Tablestore:Parallel scan

Permintaan split

splitsRequest bertipe ComputeSplitsRequest dan berisi parameter berikut.

Konfigurasi split search index

splitsRequest.splitsOptions bertipe SearchIndexSplitsOptions dan berisi parameter berikut.

Permintaan pemindaian

request bertipe ParallelScanRequest dan berisi parameter berikut.

Konfigurasi pemindaian

request.scanQuery bertipe ScanQuery dan berisi parameter berikut.

Kolom yang dikembalikan

request.columnsToGet bertipe SearchRequest.ColumnsToGet dan berisi parameter berikut.

Informasi split

computeSplits mengembalikan objek ComputeSplitsResponse yang berisi bidang-bidang berikut.

Hasil pemindaian

parallelScan mengembalikan objek ParallelScanResponse yang berisi bidang-bidang berikut.

createParallelScanIterator mengembalikan RowIterator. Iterator ini secara otomatis menggunakan nextToken untuk mengambil halaman berikutnya dan mengembalikan satu objek Row per iterasi. Iterator ini tidak mendukung pengambilan jumlah total baris yang sesuai.

Pemindaian dengan banyak worker

Contoh berikut membuat beberapa tugas pemindaian berdasarkan nilai splitsSize. Setiap tugas menggunakan currentParallelId yang unik, dan semua tugas berbagi sessionId serta maxParallel yang sama. Ukuran kolam thread tidak melebihi jumlah core CPU pada client untuk menghindari beban berlebih pada client.

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

Mengambil halaman secara manual

Contoh berikut langsung memanggil parallelScan dan meneruskan nextToken dari setiap respons ke permintaan berikutnya hingga worker saat ini telah mengambil semua baris.

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