Tablestore SDK for Java を使用して、データテーブル用の BaseData、ストリーム、または BaseAndStream トンネルを作成します。
前提条件
機能の説明
createTunnel を呼び出して、データテーブル用のトンネルを作成します。同一のデータテーブルに対して複数のトンネルを作成できます。トンネルタイプによって消費するデータが決まります。BaseData は全量データのみを消費し、Stream は増分データのみを消費し、BaseAndStream は全量データを消費した後に増分データを消費します。
データテーブルでストリームが無効になっている場合、Stream または BaseAndStream トンネルを作成すると、ストリームが自動的に有効になり、増分ログの保持期間が 7 日に設定されます。
CreateTunnelResponse createTunnel(CreateTunnelRequest request)
throws TableStoreException, ClientException次のサンプルコードは、example_table データテーブルに、example_tunnel という名前の BaseData トンネルを作成します。
String tableName = "example_table";
String tunnelName = "example_tunnel";
CreateTunnelRequest request =
new CreateTunnelRequest(
tableName, tunnelName, TunnelType.BaseData);
CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());パラメーター
CreateTunnelRequest には、次のパラメーターが含まれます：
|
名前 |
タイプ |
説明 |
|
tableName (必須) |
|
データテーブルの名前。 |
|
tunnelName (必須) |
|
トンネル名。 |
|
tunnelType (必須) |
|
トンネルタイプ。 |
|
streamTunnelConfig (オプション) |
|
|
|
streamRecordOptions (オプション) |
|
|
増分データ範囲
streamTunnelConfig は StreamTunnelConfig 型で、次のパラメーターが含まれます：
|
名前 |
タイプ |
説明 |
|
flag (オプション) |
|
|
|
startOffset (オプション) |
|
増分データの開始タイムスタンプ。単位：ミリ秒。有効範囲：[現在のシステム時刻 - ストリームの保持期間 + 5 分, 現在のシステム時刻)。このパラメーターを指定した場合、 |
|
endOffset (オプション) |
|
増分データの終了タイムスタンプ。単位：ミリ秒。両方のタイムスタンプを指定する場合、このパラメーターは |
ストリームの保持期間とは、増分ログの保持期間です。最大値は 7 日です。データテーブルでストリームを有効にする際に保持期間を設定できます。設定後は変更できません。
増分レコードの内容
streamRecordOptions は StreamRecordOptions 型で、次のパラメーターが含まれます：
|
名前 |
タイプ |
説明 |
|
getVersionGeneratorValue (オプション) |
|
増分レコードに、バージョンジェネレーターによって生成された値を含めるかどうかを指定します。デフォルト値： |
|
getSysColumns (オプション) |
|
増分レコードにシステム列を含めるかどうかを指定します。デフォルト値： |
|
getNewRowInfo (オプション) |
|
増分レコードに最新の行情報を含めるかどうかを指定します。デフォルト値： |
|
oldColumnsToGet (オプション) |
|
元の行から返す属性列。 |
|
newColumnsToGet (オプション) |
|
最新の行から返す属性列。 |
増分レコード列
streamRecordOptions.oldColumnsToGet と streamRecordOptions.newColumnsToGet は StreamColumn 型で、次のパラメーターが含まれます：
|
名前 |
タイプ |
説明 |
|
columnType (必須) |
|
属性列の選択モード。 |
|
columnNames (オプション) |
|
|
レスポンス
CreateTunnelResponse には、次のレスポンスフィールドが含まれます：
|
フィールド |
タイプ |
説明 |
|
tunnelId |
|
作成されたトンネルの ID。ID を取得するには |
シナリオ
増分データ範囲の指定
次のサンプルコードは、ストリームトンネルを作成し、増分データの範囲に直近 1 時間を指定します。
long endTime = System.currentTimeMillis() - 1_000L;
long startTime = endTime - 60 * 60 * 1000L;
StreamTunnelConfig streamConfig =
new StreamTunnelConfig(startTime, endTime);
CreateTunnelRequest request =
new CreateTunnelRequest(
"example_table",
"example_stream_tunnel",
TunnelType.Stream);
request.setStreamTunnelConfig(streamConfig);
CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());増分レコード内容の設定
次のサンプルコードは、ストリームトンネルを作成し、増分レコードにバージョンジェネレーターの値、システム列、最新の行情報、元の行の value 属性列、および最新の行のすべての属性列を返すように設定します。
StreamColumn oldColumns =
new StreamColumn(StreamColumnType.SPECIFIED_COLUMN);
oldColumns.addColumnName("value");
StreamRecordOptions recordOptions = new StreamRecordOptions();
recordOptions.setGetVersionGeneratorValue(true);
recordOptions.setGetSysColumns(true);
recordOptions.setGetNewRowInfo(true);
recordOptions.setOldColumnsToGet(oldColumns);
recordOptions.setNewColumnsToGet(
new StreamColumn(StreamColumnType.ALL_COLUMNS));
CreateTunnelRequest request =
new CreateTunnelRequest(
"example_table", "example_stream_tunnel", TunnelType.Stream);
request.setStreamRecordOptions(recordOptions);
CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());