Tablestore:並列スキャン

スプリットリクエスト

splitsRequest は ComputeSplitsRequest 型で、次のパラメーターを含みます。

検索インデックス分割構成

splitsRequest.splitsOptions は SearchIndexSplitsOptions 型で、次のパラメーターを含みます。

スキャンリクエスト

request は ParallelScanRequest 型で、次のパラメーターを含みます。

スキャン設定

request.scanQuery は ScanQuery 型で、次のパラメーターを含みます。

返される列

request.columnsToGet は SearchRequest.ColumnsToGet 型で、次のパラメーターを含みます。

スプリット情報

computeSplits は ComputeSplitsResponse を返します。これには次のフィールドが含まれます。

スキャン結果

parallelScan は ParallelScanResponse を返します。これには次のフィールドが含まれます。

createParallelScanIterator は RowIterator を返します。イテレータは自動的に nextToken を使用して後続のページを取得し、反復ごとに 1 つの 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);