Parallel scan - Tablestore SDK for Java - Tablestore - Centro de documentación de Alibaba Cloud

Puede utilizar Tablestore SDK for Java para escanear filas coincidentes en un índice de búsqueda de forma concurrente y exportar el conjunto completo de resultados cuando no se requiere un orden específico.

Requisitos previos

Instale Tablestore SDK for Java e inicialice un cliente.

El escaneo paralelo requiere Tablestore SDK for Java 5.6.0 o posterior. Para obtener información sobre las versiones, consulte Version history.

Funcionamiento

El escaneo paralelo recorre todas las filas que coinciden con una consulta dentro de un índice de búsqueda. Este proceso no garantiza el orden global de los resultados ni admite la clasificación o agregación. Si necesita resultados ordenados, agregaciones o datos de búsqueda para usuarios finales, utilice la operación Search.

Un escaneo con un solo trabajador es más sencillo de configurar. Por otro lado, un escaneo con múltiples trabajadores lee varias divisiones de forma concurrente y suele ofrecer un mayor throughput que el escaneo individual.

Un escaneo paralelo consta de los siguientes pasos:

Llame a computeSplits para obtener el paralelismo máximo splitsSize y el ID de sesión de tarea sessionId del índice de búsqueda.
Configure ParallelScanRequest. Para un escaneo con un solo trabajador, puede omitir maxParallel y currentParallelId. Para un escaneo con múltiples trabajadores, utilice la misma consulta, sessionId y maxParallel en todos los trabajadores, y especifique un valor diferente de currentParallelId para cada uno.
Llame a createParallelScanIterator para leer automáticamente todas las páginas, o llame a parallelScan y utilice nextToken para recuperar manualmente las páginas subsiguientes.
Espere a que finalicen todos los trabajadores y combine sus resultados no ordenados.

Dentro de una misma sessionId, la instantánea de datos se fija cuando se llama a parallelScan por primera vez. Las filas que se agreguen o actualicen mientras se ejecuta la tarea no se incluirán en dicha instantánea. Es posible omitir sessionId. Sin embargo, si ocurre un balanceo de carga del lado del servidor o un cambio similar durante el escaneo, los resultados podrían contener un pequeño número de filas duplicadas. Se recomienda llamar primero a computeSplits e incluir el sessionId devuelto en las solicitudes posteriores.

Importante

La sesión puede caducar anticipadamente si un cambio dinámico de esquema modifica el índice, o si ocurre un failover o balanceo de carga en el servidor. En este caso, el servidor devuelve OTSSessionExpired. Un error de red en el cliente también puede interrumpir el escaneo. Si ocurre alguno de estos errores, descarte los resultados incompletos, llame nuevamente a computeSplits y reinicie toda la tarea de escaneo desde el principio. Se pueden ejecutar hasta 10 tareas de escaneo paralelo de manera concurrente en un índice de búsqueda. Para consultar otros límites, véase Search index limits.

Llame a computeSplits para calcular las divisiones. Llame a parallelScan para recuperar páginas manualmente, o bien a createParallelScanIterator para obtener automáticamente todas las páginas.

ComputeSplitsResponse computeSplits(ComputeSplitsRequest request)
ParallelScanResponse parallelScan(ParallelScanRequest request)
RowIterator createParallelScanIterator(ParallelScanRequest request)

El siguiente ejemplo escanea todas las filas mediante un solo trabajador y devuelve los campos category y price. El objeto RowIterator recupera automáticamente las páginas subsiguientes.

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

ComputeSplitsRequest splitsRequest = ComputeSplitsRequest.newBuilder()
        .tableName(tableName)
        .splitsOptions(new SearchIndexSplitsOptions(indexName))
        .build();
ComputeSplitsResponse splitsResponse =
        client.computeSplits(splitsRequest);

ScanQuery scanQuery = ScanQuery.newBuilder()
        .query(QueryBuilders.matchAll())
        .limit(2000)
        .build();
ParallelScanRequest request = ParallelScanRequest.newBuilder()
        .tableName(tableName)
        .indexName(indexName)
        .scanQuery(scanQuery)
        .addColumnsToGet("category", "price")
        .sessionId(splitsResponse.getSessionId())
        .build();

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

Parámetros

Solicitud de división

splitsRequest es de tipo ComputeSplitsRequest y contiene los siguientes parámetros.

Nombre	Tipo	Descripción
tableName (obligatorio)	String	Nombre de la tabla de datos.
splitsOptions (obligatorio)	SplitsOptions	Configuración de la división. Para un índice de búsqueda, establezca este parámetro como `SearchIndexSplitsOptions`.

Configuración de división del índice de búsqueda

splitsRequest.splitsOptions es de tipo SearchIndexSplitsOptions y contiene el siguiente parámetro.

Nombre	Tipo	Descripción
indexName (obligatorio)	String	Nombre del índice de búsqueda.

Solicitud de escaneo

request es de tipo ParallelScanRequest y contiene los siguientes parámetros.

Nombre	Tipo	Descripción
tableName (obligatorio)	String	Nombre de la tabla de datos.
indexName (obligatorio)	String	Nombre del índice de búsqueda.
scanQuery (obligatorio)	ScanQuery	Condición de escaneo, número de filas devueltas por solicitud y configuración de paralelismo.
columnsToGet (opcional)	SearchRequest.ColumnsToGet	Columnas que se devolverán. Si no se especifica este parámetro, solo se devuelven las columnas de clave principal.
sessionId (opcional)	`byte[]`	ID de sesión de tarea devuelto por `computeSplits`. Se recomienda especificar este parámetro para utilizar la misma instantánea de datos durante todo el escaneo.
timeoutInMillisecond (opcional)	int	Periodo de tiempo de espera a nivel de solicitud, en milisegundos. El valor predeterminado es `-1`, lo cual indica que no se establece ningún tiempo de espera a nivel de solicitud.

Configuración del escaneo

request.scanQuery es de tipo ScanQuery y contiene los siguientes parámetros.

Nombre	Tipo	Descripción
query (obligatorio)	Query	Condición de consulta que define el alcance del escaneo. El escaneo paralelo admite consultas de término, coincidencia, rango, geográficas, anidadas, entre otras. Configure la consulta de la misma manera que para la operación Search. Para escanear todas las filas del índice de búsqueda, establezca el tipo de consulta como `MatchAllQuery`.
limit (opcional)	Integer	Número máximo de filas devueltas por solicitud. Valor predeterminado: 2000. Se recomienda utilizar el valor predeterminado.
maxParallel (opcional)	Integer	Paralelismo de la tarea de escaneo. El valor no puede exceder `ComputeSplitsResponse.splitsSize`. Valor predeterminado: 1.
currentParallelId (opcional)	Integer	ID del trabajador actual. Este parámetro es obligatorio si `maxParallel` es mayor que 1. Asigne a cada trabajador un valor único en el rango `[0, maxParallel)`.
aliveTime (opcional)	Integer	Intervalo máximo entre dos solicitudes de página para la tarea de escaneo, en segundos. Valores válidos: de 1 a 600. Valor predeterminado: 60. El periodo de validez se actualiza cada vez que se recuperan filas correctamente.
token (opcional)	`byte[]`	Token de paginación. Omita este parámetro en la primera solicitud. Para la paginación manual, establézcalo como `nextToken` de la respuesta anterior. El SDK gestiona este parámetro cuando utiliza `createParallelScanIterator`.

Nota

El valor máximo de limit admitido por el servidor es 10000. Se recomienda no establecer limit en este valor.

Columnas a devolver

request.columnsToGet es de tipo SearchRequest.ColumnsToGet y contiene los siguientes parámetros.

Nombre	Tipo	Descripción
columns (opcional)	`List<String>`	Nombres de los campos del índice de búsqueda que se devolverán. No se puede devolver un campo que exista solo en la tabla de datos pero no esté incluido en el índice de búsqueda. Se pueden devolver campos de tipo Date, Geo-point, IP, Vector, JSON/Nested y arrays.
returnAllFromIndex (opcional)	boolean	Especifica si se deben devolver todos los campos del índice de búsqueda. Valor predeterminado: `false`. Si establece este parámetro en `true`, no es necesario especificar `columns`.
returnAll (opcional)	boolean	El escaneo paralelo no admite este parámetro. No lo establezca en `true`.

Valores de retorno

Información de división

computeSplits devuelve ComputeSplitsResponse, que contiene los siguientes campos.

Nombre	Tipo	Descripción
sessionId	`byte[]`	ID de sesión de tarea que se utiliza para escanear filas en la misma instantánea de datos.
splitsSize	Integer	Paralelismo máximo admitido por el índice de búsqueda.

Resultados del escaneo

parallelScan devuelve ParallelScanResponse, que contiene los siguientes campos.

Nombre	Tipo	Descripción
rows	`List<Row>`	Filas devueltas en la respuesta actual.
nextToken	`byte[]`	Token para la siguiente página. Si este valor es `null`, significa que el trabajador actual ha recuperado todas las filas.
bodyBytes	long	Tamaño del cuerpo de la respuesta en bytes.

createParallelScanIterator devuelve RowIterator. El iterador utiliza automáticamente nextToken para recuperar las páginas subsiguientes y devuelve una Row por iteración. No admite la recuperación del número total de filas coincidentes.

Ejemplos

Escaneo con múltiples trabajadores

El siguiente ejemplo crea múltiples tareas de escaneo basadas en splitsSize. Cada tarea utiliza un currentParallelId único, y todas comparten el mismo sessionId y maxParallel. El pool de hilos no excede la cantidad de núcleos de CPU del cliente para evitar una carga excesiva.

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

Recuperación manual de páginas

El siguiente ejemplo llama directamente a parallelScan y pasa nextToken de cada respuesta a la siguiente solicitud hasta que el trabajador actual haya recuperado todas las filas.

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