Tablestore:増分データのコンシューム

前提条件

• Tablestore SDK for Java がインストールされ、クライアントが初期化されていること。

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

仕組み

ストリームは、テーブルからの増分変更をシャードに整理します。ストリームをコンシュームするには、4 つの API を順番に呼び出します。ストリームのリスト取得、ストリームの詳細取得、シャードイテレーターの取得、レコードのフェッチです。

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

主な手順：

1. listStream(ListStreamRequest) を呼び出して、インスタンス内でストリームが有効化されたすべてのテーブルの streamId 値のリストを取得します。

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

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

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

次の例では、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("シャード数: " + 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.size());
    for (StreamRecord record : records) {
        System.out.println("レコードタイプ: " + record.getRecordType()
                + ", プライマリキー: " + record.getPrimaryKey());
    }

    // nextShardIterator を使用して、後続の増分レコードのフェッチを継続します。
    System.out.println("次のイテレーター: "
            + (recordResponse.getNextShardIterator() != null ? "あり" : "なし"));
}

パラメータ

例

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

    // null の nextShardId は、すべてのシャードが走査されたことを示します。
    if (response.getNextShardId() == null) {
        break;
    }
    startShardId = response.getNextShardId();
}
System.out.println("合計シャード数: " + 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("ポーリングされた合計レコード数: " + totalRecords);

関連ドキュメント

• トンネルサービス