Tablestore：並發匯出資料

計算分區請求

splitsRequest 的類型為 ComputeSplitsRequest，包含以下參數。

索引分區配置

splitsRequest.splitsOptions 的類型為 SearchIndexSplitsOptions，包含以下參數。

掃描請求

request 的類型為 ParallelScanRequest，包含以下參數。

掃描配置

request.scanQuery 的類型為 ScanQuery，包含以下參數。

返回列配置

request.columnsToGet 的類型為 SearchRequest.ColumnsToGet，包含以下參數。

分區資訊

computeSplits 返回 ComputeSplitsResponse，包含以下欄位。

掃描結果

parallelScan 返回 ParallelScanResponse，包含以下欄位。

createParallelScanIterator 返回 RowIterator。該迭代器自動使用 nextToken 擷取後續分頁，每次迭代返回一個 Row，不支援擷取匹配總行數。

多並發掃描

以下樣本根據 splitsSize 建立多個掃描任務。每個任務使用唯一的 currentParallelId，全部任務共用同一 sessionId 和 maxParallel。線程池大小不超過用戶端的 CPU 核心數，避免同時運行過多線程增加用戶端負載。

String tableName = "example_table";
String indexName = "example_index";

ComputeSplitsResponse splitsResponse = client.computeSplits(
        ComputeSplitsRequest.newBuilder()
                .tableName(tableName)
                .splitsOptions(new SearchIndexSplitsOptions(indexName))
                .build());
int maxParallel = splitsResponse.getSplitsSize();
int workerCount = Math.min(
        maxParallel, Runtime.getRuntime().availableProcessors());

ExecutorService executor = Executors.newFixedThreadPool(workerCount);
List<Future<Integer>> futures = new ArrayList<Future<Integer>>();
try {
    for (int parallelId = 0; parallelId < maxParallel; parallelId++) {
        final int currentParallelId = parallelId;
        futures.add(executor.submit(new Callable<Integer>() {
            @Override
            public Integer call() {
                ScanQuery scanQuery = ScanQuery.newBuilder()
                        .query(QueryBuilders.matchAll())
                        .limit(2000)
                        .maxParallel(maxParallel)
                        .currentParallelId(currentParallelId)
                        .build();
                ParallelScanRequest request =
                        ParallelScanRequest.newBuilder()
                                .tableName(tableName)
                                .indexName(indexName)
                                .scanQuery(scanQuery)
                                .returnAllColumnsFromIndex(true)
                                .sessionId(splitsResponse.getSessionId())
                                .build();

                int rowCount = 0;
                RowIterator iterator =
                        client.createParallelScanIterator(request);
                while (iterator.hasNext()) {
                    Row row = iterator.next();
                    System.out.println(row);
                    rowCount++;
                }
                return rowCount;
            }
        }));
    }

    long totalRows = 0;
    for (Future<Integer> future : futures) {
        totalRows += future.get();
    }
    System.out.println("Total rows: " + totalRows);
} finally {
    executor.shutdown();
}

手動翻頁

以下樣本直接調用 parallelScan，並將每次響應的 nextToken 寫入下一次請求，直到當前並發任務讀取完畢。

String tableName = "example_table";
String indexName = "example_index";

ComputeSplitsResponse splitsResponse = client.computeSplits(
        ComputeSplitsRequest.newBuilder()
                .tableName(tableName)
                .splitsOptions(new SearchIndexSplitsOptions(indexName))
                .build());
ScanQuery scanQuery = ScanQuery.newBuilder()
        .query(QueryBuilders.matchAll())
        .limit(2000)
        .maxParallel(1)
        .currentParallelId(0)
        .build();
ParallelScanRequest request = ParallelScanRequest.newBuilder()
        .tableName(tableName)
        .indexName(indexName)
        .scanQuery(scanQuery)
        .addColumnsToGet("category", "price")
        .sessionId(splitsResponse.getSessionId())
        .build();

long totalRows = 0;
do {
    ParallelScanResponse response = client.parallelScan(request);
    for (Row row : response.getRows()) {
        System.out.println(row);
        totalRows++;
    }
    scanQuery.setToken(response.getNextToken());
} while (scanQuery.getToken() != null);
System.out.println("Total rows: " + totalRows);