Java SDK 在巨量資料離線情境下將多行資料批量寫入資料表,支援插入、修改、刪除三種行操作。
前提條件
安裝 Tablestore Java SDK並初始化用戶端。
功能說明
public BulkImportResponse bulkImport(BulkImportRequest bulkImportRequest) throws TableStoreException, ClientException
在單個請求中提交多行資料操作,由服務端按行獨立寫入並按行返回結果,適用於巨量資料情境下的離線大量匯入。
-
通過
addRowChange(RowChange)或addRowChanges(List<RowChange>)添加行操作;支援RowPutChange(插入)、RowUpdateChange(修改)、RowDeleteChange(刪除)三種子類,可在同一請求中混合使用。 -
通過
BulkImportResponse.getResult(succeedRows, failedRows)拿成功和失敗的行;isAllSucceed()判斷全量成功。每行結果獨立,部分行失敗不影響其他行寫入。 -
BulkImportRequest.createRequestForRetry(failedRows)基於失敗行產生只包含失敗行的重試請求,用於按行層級重試。
以下樣本將 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> |
行巨集指令清單。每個元素為 |
情境樣本
單請求混合三種行操作
在同一個 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());
}