Table Store Java SDK 離線寫入資料 - Tablestore

前提條件

安裝 Tablestore Java SDK並初始化用戶端。

功能說明

調用 bulkImport 在單個請求中提交多行資料操作。服務端按行獨立處理並返回行級結果，部分行失敗不影響其他行，適用於巨量資料情境下的離線大量匯入。

public BulkImportResponse bulkImport(BulkImportRequest bulkImportRequest) throws TableStoreException, ClientException

以下樣本將 5 行資料批量插入資料表 bulk_import_demo，並列印成功和失敗行數。

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>

行巨集指令清單，可在同一請求中混合以下操作。

RowPutChange：寫入一行資料。具體配置請參見寫入單行資料。
RowUpdateChange：更新一行資料。具體配置請參見更新單行資料。
RowDeleteChange：刪除一行資料。具體配置請參見刪除單行資料。

傳回值

匯入結果

BulkImportResponse 包含以下業務欄位。

名稱	類型	說明
rowResults	List<BulkImportResponse.RowResult>	各行操作的結果，順序與請求中的 `rowChanges` 一致。通過 `getRowResults()` 擷取。

行級結果

BulkImportResponse.rowResults[] 中每個元素的類型為 BulkImportResponse.RowResult。

名稱	類型	說明
succeed	boolean	該行操作是否成功。通過 `isSucceed()` 擷取。
error	Error	該行操作失敗時的錯誤資訊；成功時為 `null`。
index	int	該行操作在請求的 `rowChanges` 中的索引。

情境樣本

單請求混合三種行操作

在同一個 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());
}