Tablestore:Analyse parallèle

Requête de partitionnement

Le paramètre splitsRequest est de type ComputeSplitsRequest et contient les paramètres suivants.

Configuration du partitionnement de l'index de recherche

Le paramètre splitsRequest.splitsOptions est de type SearchIndexSplitsOptions et contient le paramètre suivant.

Requête d'analyse

Le paramètre request est de type ParallelScanRequest et contient les paramètres suivants.

Configuration de l'analyse

Le paramètre request.scanQuery est de type ScanQuery et contient les paramètres suivants.

Colonnes à renvoyer

Le paramètre request.columnsToGet est de type SearchRequest.ColumnsToGet et contient les paramètres suivants.

Informations sur le partitionnement

La méthode computeSplits renvoie un objet ComputeSplitsResponse, qui contient les champs suivants.

Résultats de l'analyse

La méthode parallelScan renvoie un objet ParallelScanResponse, qui contient les champs suivants.

La méthode createParallelScanIterator renvoie un objet RowIterator. L'itérateur utilise automatiquement nextToken pour récupérer les pages suivantes et renvoie une Row par itération. Il ne prend pas en charge la récupération du nombre total de lignes correspondantes.

Analyse avec plusieurs workers

L'exemple suivant crée plusieurs tâches d'analyse basées sur la valeur splitsSize. Chaque tâche utilise un currentParallelId unique, et toutes les tâches partagent le même sessionId et la même valeur maxParallel. Le pool de threads ne dépasse pas le nombre de cœurs CPU du client afin d'éviter une charge excessive côté 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();
}

Récupération manuelle des pages

L'exemple suivant appelle directement parallelScan et transmet la valeur nextToken de chaque réponse à la requête suivante jusqu'à ce que le worker actuel ait récupéré toutes les lignes.

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