オフラインデータのバルクインポート - Tablestore SDK for Java - Tablestore

前提条件

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

説明

bulkImport を呼び出して、1 回のリクエストで複数の行操作を送信します。サーバーは各行を個別に処理し、行ごとの結果を返します。そのため、1 つの行で障害が発生しても、他の行の操作には影響しません。このメソッドは、ビッグデータシナリオでのオフラインのバルクインポートを目的としています。

public BulkImportResponse bulkImport(BulkImportRequest bulkImportRequest) throws TableStoreException, ClientException

次の例では、1 回のバルクリクエストで bulk_import_demo テーブルに 5 行を挿入し、成功した行と失敗した行の数を表示します。

String tableName = "bulk_import_demo";

BulkImportRequest request = new BulkImportRequest(tableName);

List<RowChange> rowChanges = new ArrayList<RowChange>();
for (int i = 0; i < 5; i++) {
    PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
            .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("row" + i))
            .build();
    RowPutChange put = new RowPutChange(tableName, pk);
    put.addColumn(new Column("col1", ColumnValue.fromString("v" + i)));
    rowChanges.add(put);
}
request.addRowChanges(rowChanges);

BulkImportResponse response = client.bulkImport(request);

// 行ごとの結果を succeedRows と failedRows に取得
List<BulkImportResponse.RowResult> succeedRows = new ArrayList<BulkImportResponse.RowResult>();
List<BulkImportResponse.RowResult> failedRows = new ArrayList<BulkImportResponse.RowResult>();
response.getResult(succeedRows, failedRows);

System.out.println("All succeed: " + response.isAllSucceed());
System.out.println("Succeed: " + succeedRows.size() + ", Failed: " + failedRows.size());

パラメーター

名前

タイプ

説明

tableName (必須)

String

テーブルの名前。

rowChanges (必須)

List<RowChange>

行操作のリスト。1 回のリクエストで次の操作を混在させることができます：

RowPutChange：行を書き込みます。設定の詳細については、「行の書き込み」をご参照ください。
RowUpdateChange：行を更新します。設定の詳細については、「行の更新」をご参照ください。
RowDeleteChange：行を削除します。設定の詳細については、「行の削除」をご参照ください。

レスポンス

インポート結果

BulkImportResponse には、次の操作固有のフィールドが含まれています。

名前	タイプ	説明
rowResults	List<BulkImportResponse.RowResult>	各行の操作結果のリストです。結果はリクエスト内の `rowChanges` と同じ順序で返されます。成功した行と失敗した行を振り分けるには、`getResult` メソッドを使用します。

行の結果

BulkImportResponse.rowResults[] の各要素は BulkImportResponse.RowResult 型です。

名前	タイプ	説明
succeed	boolean	行操作が成功したかどうかを示します。`isSucceed()` を呼び出して値を取得します。
error	Error	失敗した行操作のエラー。操作が成功した場合、値は `null` です。
index	int	リクエストの `rowChanges` 内での行操作のインデックス。

例

1 つのリクエストでの書き込み、更新、削除の組み合わせ

1 つの BulkImportRequest を使用して、書き込み、更新、削除の操作をまとめて送信します。

String tableName = "bulk_import_demo";

BulkImportRequest request = new BulkImportRequest(tableName);

// 新しい行を挿入
PrimaryKey pkPut = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("mixed_put"))
        .build();
RowPutChange put = new RowPutChange(tableName, pkPut);
put.addColumn(new Column("col1", ColumnValue.fromString("put_value")));
request.addRowChange(put);

// 既存の行を更新 (カラムを追加)
PrimaryKey pkUpdate = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("row0"))
        .build();
RowUpdateChange update = new RowUpdateChange(tableName, pkUpdate);
update.put(new Column("col2", ColumnValue.fromLong(100)));
request.addRowChange(update);

// 既存の行を削除
PrimaryKey pkDelete = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("row1"))
        .build();
RowDeleteChange delete = new RowDeleteChange(tableName, pkDelete);
request.addRowChange(delete);

BulkImportResponse response = client.bulkImport(request);
System.out.println("Mixed all succeed: " + response.isAllSucceed());

失敗した行のリトライ

createRequestForRetry(failedRows) を使用して、失敗した行から新しいリクエストを構築します。リトライリクエストには失敗した行のみが含まれるため、すでに成功した行は再適用されません。

String tableName = "bulk_import_demo";

BulkImportRequest request = new BulkImportRequest(tableName);

PrimaryKey pk = PrimaryKeyBuilder.createPrimaryKeyBuilder()
        .addPrimaryKeyColumn("pk", PrimaryKeyValue.fromString("good"))
        .build();
RowPutChange row = new RowPutChange(tableName, pk);
row.addColumn(new Column("col1", ColumnValue.fromString("ok")));
request.addRowChange(row);

BulkImportResponse response = client.bulkImport(request);

List<BulkImportResponse.RowResult> succeedRows = new ArrayList<BulkImportResponse.RowResult>();
List<BulkImportResponse.RowResult> failedRows = new ArrayList<BulkImportResponse.RowResult>();
response.getResult(succeedRows, failedRows);

// いずれかの行が失敗した場合、失敗した行のみを含むリトライリクエストを構築
if (!failedRows.isEmpty()) {
    BulkImportRequest retryRequest = request.createRequestForRetry(failedRows);
    BulkImportResponse retryResponse = client.bulkImport(retryRequest);
    System.out.println("Retry all succeed: " + retryResponse.isAllSucceed());
}