Tablestore:オフラインデータのバルクインポート

前提条件

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

説明

public BulkImportResponse bulkImport(BulkImportRequest bulkImportRequest) throws TableStoreException, ClientException

bulkImport は、単一のリクエストで複数の行操作を送信します。サーバーは各行を独立して処理し、行ごとの結果を返します。このメソッドは、ビッグデータシナリオにおけるオフラインのバルクインポートに適しています。

• addRowChange(RowChange) または addRowChanges(List<RowChange>) を使用して行操作を追加します。どちらのメソッドも、RowChange の 3 つのサブクラス (RowPutChange、RowUpdateChange、RowDeleteChange) を受け取ることができます。単一のリクエストで、3 つのサブクラスすべてを組み合わせることができます。

• BulkImportResponse.getResult(succeedRows, failedRows) を呼び出して、2 つのリストに行ごとの結果を格納します。isAllSucceed() を呼び出して、すべての行が成功したかどうかを確認します。各行の結果は独立しており、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());

パラメータ

例

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

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

関連ドキュメント

• バッチ行操作

• オフラインデータのバルクエクスポート