Tablestore:Mengonsumsi data inkremental

Prasyarat

• Instal Tablestore SDK for Java dan inisialisasi klien.

• Aktifkan Stream pada tabel dengan menetapkan StreamSpecification saat pembuatan tabel. Untuk detailnya, lihat Buat tabel data.

Deskripsi

Stream mengorganisasi perubahan inkremental dari suatu tabel ke dalam shard. Untuk mengonsumsi stream, panggil empat API secara berurutan: list streams, describe a stream, get a shard iterator, dan fetch records.

1. Panggil listStream(ListStreamRequest) untuk mendaftar nilai streamId dari semua tabel yang diaktifkan Stream di instans tersebut.

2. Panggil describeStream(DescribeStreamRequest) untuk mengambil metadata stream (waktu pembuatan, waktu kedaluwarsa, status saat ini) dan daftar objek Shard.

3. Panggil getShardIterator(GetShardIteratorRequest) untuk mendapatkan iterator baca (shardIterator) untuk Shard tertentu. Iterator ini menandai posisi awal pengambilan catatan inkremental.

4. Panggil getStreamRecord(GetStreamRecordRequest) dengan shardIterator untuk mengambil batch catatan inkremental (daftar objek StreamRecord). Gunakan nextShardIterator yang dikembalikan untuk mengambil catatan berikutnya.

public ListStreamResponse listStream(ListStreamRequest request) throws TableStoreException, ClientException
public DescribeStreamResponse describeStream(DescribeStreamRequest request) throws TableStoreException, ClientException
public GetShardIteratorResponse getShardIterator(GetShardIteratorRequest request) throws TableStoreException, ClientException
public GetStreamRecordResponse getStreamRecord(GetStreamRecordRequest request) throws TableStoreException, ClientException

Contoh berikut mengonsumsi stream stream_test_demo secara end-to-end dan mencetak tipe serta kunci primer setiap catatan.

String demoTable = "stream_test_demo";

// 1. Daftar semua tabel di instans yang diaktifkan Stream dan temukan streamId tabel target.
ListStreamRequest listRequest = new ListStreamRequest(demoTable);
ListStreamResponse listResponse = client.listStream(listRequest);

String targetStreamId = null;
for (Stream stream : listResponse.getStreams()) {
    if (demoTable.equals(stream.getTableName())) {
        targetStreamId = stream.getStreamId();
        break;
    }
}
System.out.println("Stream ID: " + targetStreamId);

// 2. Kueri semua shard dari Stream.
DescribeStreamRequest describeRequest = new DescribeStreamRequest(targetStreamId);
DescribeStreamResponse describeResponse = client.describeStream(describeRequest);
List<StreamShard> shards = describeResponse.getShards();
System.out.println("Jumlah shard: " + shards.size());

if (!shards.isEmpty()) {
    String shardId = shards.get(0).getShardId();

    // 3. Dapatkan iterator baca awal dari shard.
    GetShardIteratorRequest iterRequest =
            new GetShardIteratorRequest(targetStreamId, shardId);
    GetShardIteratorResponse iterResponse = client.getShardIterator(iterRequest);
    String shardIterator = iterResponse.getShardIterator();

    // 4. Gunakan iterator untuk menarik catatan inkremental dari shard.
    GetStreamRecordRequest recordRequest = new GetStreamRecordRequest(shardIterator);
    recordRequest.setLimit(100);
    GetStreamRecordResponse recordResponse = client.getStreamRecord(recordRequest);

    List<StreamRecord> records = recordResponse.getRecords();
    System.out.println("Catatan yang diambil: " + records.size());
    for (StreamRecord record : records) {
        System.out.println("RecordType: " + record.getRecordType()
                + ", PK: " + record.getPrimaryKey());
    }

    // nextShardIterator digunakan untuk melanjutkan penarikan catatan inkremental berikutnya.
    System.out.println("Iterator berikutnya: "
            + (recordResponse.getNextShardIterator() != null ? "ya" : "tidak"));
}

Parameter

Respons

Contoh

String currentStreamId = "<your-stream-id>";
String startShardId = null;
int totalShards = 0;

while (true) {
    DescribeStreamRequest request = new DescribeStreamRequest(currentStreamId);
    if (startShardId != null) {
        request.setInclusiveStartShardId(startShardId);
    }
    request.setShardLimit(50);

    DescribeStreamResponse response = client.describeStream(request);
    totalShards += response.getShards().size();

    // Nilai nextShardId null menunjukkan bahwa semua shard telah dilalui.
    if (response.getNextShardId() == null) {
        break;
    }
    startShardId = response.getNextShardId();
}
System.out.println("Total shard: " + totalShards);

String currentStreamId = "<your-stream-id>";
String shardId = "<your-shard-id>";

GetShardIteratorRequest iterRequest =
        new GetShardIteratorRequest(currentStreamId, shardId);
String shardIterator = client.getShardIterator(iterRequest).getShardIterator();

int totalRecords = 0;
while (shardIterator != null) {
    GetStreamRecordRequest recordRequest = new GetStreamRecordRequest(shardIterator);
    recordRequest.setLimit(100);
    GetStreamRecordResponse response = client.getStreamRecord(recordRequest);

    totalRecords += response.getRecords().size();
    shardIterator = response.getNextShardIterator();
}
System.out.println("Total catatan dari polling: " + totalRecords);