Tablestore:増分データの消費

前提条件

• Tablestore SDK for Java をインストールし、クライアントを初期化しておきます。

• テーブルのストリームは、テーブル作成時に StreamSpecification を設定することで有効になります。詳細については、「データテーブルを作成する」をご参照ください。

説明

ストリームは、テーブルからの増分データをシャードに整理します。ストリームを消費するには、ストリームのリスト表示、ストリームの説明、シャードイテレーターの取得、レコードのフェッチという 4 つの API を順番に呼び出します。

1. listStream(ListStreamRequest) を呼び出して、インスタンス内のすべての Stream が有効化されたテーブルの streamId 値を一覧表示します。

2. describeStream(DescribeStreamRequest) を呼び出して、ストリームのメタデータ (作成時刻、有効期限、現在のステータス) とShard オブジェクトのリストを取得します。

3. getShardIterator(GetShardIteratorRequest) を呼び出して、特定の Shard の読み取りイテレーター (shardIterator) を取得します。 イテレーターは、増分レコードのフェッチを開始する場所を示します。

4. shardIterator を使用して getStreamRecord(GetStreamRecordRequest) を呼び出し、増分レコードのバッチ (StreamRecord オブジェクトのリスト) を取得します。返された nextShardIterator を使用して後続のレコードを取得します。

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

以下の例では、stream_test_demo ストリームをエンドツーエンドで消費し、各レコードのタイプとプライマリキーを出力します。

String demoTable = "stream_test_demo";

// 1. インスタンス内でストリームが有効なすべてのテーブルをリスト表示し、ターゲットテーブルの streamId を見つけます。
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. ストリームのすべてのシャードをクエリします。
DescribeStreamRequest describeRequest = new DescribeStreamRequest(targetStreamId);
DescribeStreamResponse describeResponse = client.describeStream(describeRequest);
List<StreamShard> shards = describeResponse.getShards();
System.out.println("Shard count: " + shards.size());

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

    // 3. シャードの初期読み取りイテレーターを取得します。
    GetShardIteratorRequest iterRequest =
            new GetShardIteratorRequest(targetStreamId, shardId);
    GetShardIteratorResponse iterResponse = client.getShardIterator(iterRequest);
    String shardIterator = iterResponse.getShardIterator();

    // 4. イテレーターを使用して、シャードから増分レコードをプルします。
    GetStreamRecordRequest recordRequest = new GetStreamRecordRequest(shardIterator);
    recordRequest.setLimit(100);
    GetStreamRecordResponse recordResponse = client.getStreamRecord(recordRequest);

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

    // nextShardIterator を使用して、後続の増分レコードをプルし続けます。
    System.out.println("Next iterator: "
            + (recordResponse.getNextShardIterator() != null ? "yes" : "no"));
}

パラメーター

レスポンス

例

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();

    // nextShardId が null の場合、すべてのシャードが走査されたことを示します。
    if (response.getNextShardId() == null) {
        break;
    }
    startShardId = response.getNextShardId();
}
System.out.println("Total shards: " + 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("Polling total records: " + totalRecords);