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

Utilice Tablestore SDK for Python para escanear datos coincidentes en un índice de búsqueda de forma paralela y exportar un conjunto de resultados completo sin ordenar.

Requisitos previos

Instale Tablestore SDK for Python e inicialice un cliente.

Descripción

El escaneo paralelo exporta todas las filas que coinciden con una consulta en un índice de búsqueda. Los resultados no tienen un orden global y no se admite la clasificación ni la agregación. Para ordenar o agregar resultados, o bien para devolver resultados de búsqueda a los usuarios finales, utilice la API Search. Un solo worker es más sencillo de configurar, mientras que varios workers leen múltiples splits simultáneamente y suelen ofrecer un mayor throughput.

El flujo de trabajo es el siguiente: llame a compute_splits para obtener la concurrencia máxima splits_size y el identificador de sesión session_id. Configure cada worker con la misma consulta, session_id y max_parallel, además de un current_parallel_id único. Cada worker llama a parallel_scan y utiliza next_token para leer las páginas siguientes. Finalmente, espere a que terminen todos los workers y combine sus resultados sin ordenar.

Importante

Los workers bajo el mismo session_id establecen una instantánea de datos cuando comienza el primer escaneo. Las actualizaciones dinámicas del esquema, el failover o el balanceo de carga pueden hacer que una sesión caduque antes de tiempo y devuelva OTSSessionExpired; los fallos de red también pueden interrumpir un escaneo. En estos casos, descarte los resultados incompletos, llame nuevamente a compute_splits y reinicie todos los workers desde el principio. Se pueden ejecutar hasta 10 trabajos de escaneo paralelo simultáneamente en un mismo índice de búsqueda.

A continuación se muestra un ejemplo que calcula los splits y luego escanea todos los datos mediante un solo worker.

splits = client.compute_splits("example_table", "example_index")
next_token = None
rows = []

while True:
    scan_query = ScanQuery(
        MatchAllQuery(),
        limit=2000,
        next_token=next_token,
        current_parallel_id=0,
        max_parallel=1,
        alive_time=60,
    )
    response = client.parallel_scan(
        "example_table",
        "example_index",
        scan_query,
        splits.session_id,
        ColumnsToGet(return_type=ColumnReturnType.ALL_FROM_INDEX),
    )
    rows.extend(response.rows)
    next_token = response.next_token
    if not next_token:
        break

print(len(rows))

Parámetros

Cálculo de splits

compute_splits(table_name, index_name) contiene los siguientes parámetros.

Nombre	Tipo	Descripción
table_name (obligatorio)	`str`	El nombre de la tabla de datos.
index_name (obligatorio)	`str`	El nombre del índice de búsqueda.

Solicitud de escaneo

parallel_scan contiene los siguientes parámetros.

Nombre	Tipo	Descripción
table_name (obligatorio)	`str`	El nombre de la tabla de datos.
index_name (obligatorio)	`str`	El nombre del índice de búsqueda.
scan_query (obligatorio)	`ScanQuery`	La condición de escaneo, la paginación y las configuraciones de concurrencia.
session_id (obligatorio)	`bytes`	El identificador de sesión devuelto por `compute_splits`, utilizado para mantener la misma instantánea de datos.
columns_to_get (opcional)	`ColumnsToGet`	La configuración de columnas a devolver. Si se omite, solo se devuelven las columnas de clave primaria.
timeout_s (opcional)	`int`	El tiempo de espera de la solicitud en segundos.

Configuración del escaneo

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

Nombre	Tipo	Descripción
query (obligatorio)	`Query`	La condición de consulta que define el alcance del escaneo. Utilice `MatchAllQuery` para escanear todas las filas.
limit (obligatorio)	`int`	El número máximo de filas por solicitud. Se recomienda el valor predeterminado `2000`. El servidor permite hasta `10000`, pero no se recomienda utilizar el máximo.
next_token (obligatorio)	`bytes`	El token de paginación. Establézcalo en `None` en la primera solicitud y utilice `next_token` de la respuesta anterior en las solicitudes subsiguientes.
current_parallel_id (obligatorio)	`int`	El ID del worker actual. Valores válidos: `[0, max_parallel)`. Cada worker debe usar un valor único.
max_parallel (obligatorio)	`int`	La concurrencia del trabajo, que no puede exceder `splits_size`.
alive_time (opcional)	`int`	El período máximo de validez entre dos solicitudes de página, en segundos. Valores válidos: de 1 a 600. Valor predeterminado: `60`. El período se actualiza tras una respuesta exitosa.

Columnas de retorno

columns_to_get es de tipo ColumnsToGet y contiene los siguientes parámetros.

Nombre	Tipo	Descripción
column_names (opcional)	`list[str]`	Los nombres de los campos del índice de búsqueda que se deben devolver. Especifique este parámetro solo cuando `return_type` sea `SPECIFIED`.
return_type (opcional)	`ColumnReturnType`	El modo de retorno de columnas. El escaneo paralelo admite `NONE`, `SPECIFIED` y `ALL_FROM_INDEX`, pero no `ALL`.

Respuesta

Información de splits

compute_splits devuelve información sobre los splits.

Campo	Tipo	Descripción
session_id	`bytes`	El identificador de sesión del trabajo.
splits_size	`int`	La concurrencia máxima admitida por el índice de búsqueda.

Resultado del escaneo

parallel_scan devuelve un resultado de escaneo.

Campo	Tipo	Descripción
rows	`list[Row]`	Las filas devueltas por la solicitud.
next_token	`bytes`	El token para la siguiente página. Un valor vacío indica que el worker actual ha terminado.

Respuestas compatibles con tuplas

El escaneo paralelo se admite a partir de Tablestore SDK for Python 5.2.0, que devuelve objetos de respuesta. En la versión 5.2.1 y posteriores, puede llamar a ComputeSplitsResponse.v1_response() y ParallelScanResponse.v1_response() para obtener tuplas. Para código nuevo, acceda directamente a los atributos del objeto de respuesta.

session_id, splits_size = splits.v1_response()
rows, next_token = response.v1_response()

Ejemplos

Escaneo con múltiples workers

El siguiente ejemplo crea workers basados en splits_size. El pool de hilos no excede el número de núcleos de CPU del cliente, y todos los workers comparten la sesión y la concurrencia máxima.

from concurrent.futures import ThreadPoolExecutor
import os

def scan_split(parallel_id, max_parallel, session_id):
    rows = []
    next_token = None
    while True:
        scan_query = ScanQuery(
            MatchAllQuery(),
            limit=2000,
            next_token=next_token,
            current_parallel_id=parallel_id,
            max_parallel=max_parallel,
            alive_time=60,
        )
        response = client.parallel_scan(
            "example_table",
            "example_index",
            scan_query,
            session_id,
            ColumnsToGet(return_type=ColumnReturnType.ALL_FROM_INDEX),
        )
        rows.extend(response.rows)
        next_token = response.next_token
        if not next_token:
            return rows

splits = client.compute_splits("example_table", "example_index")
worker_count = min(splits.splits_size, os.cpu_count() or 1)
with ThreadPoolExecutor(max_workers=worker_count) as executor:
    futures = [
        executor.submit(
            scan_split,
            parallel_id,
            splits.splits_size,
            splits.session_id,
        )
        for parallel_id in range(splits.splits_size)
    ]
    all_rows = [row for future in futures for row in future.result()]

print(len(all_rows))