Escaneo paralelo - Tablestore - Centro de documentación de Alibaba Cloud

Use el Tablestore SDK for Go para dividir los datos del índice de búsqueda y escanear las divisiones en paralelo; esto permite una exportación eficiente a gran escala.

Requisitos previos

Antes de empezar, complete las siguientes preparaciones:

Instale Tablestore Go SDK e inicialice un cliente.
El escaneo paralelo requiere la versión 1.6.0 o posterior del Tablestore SDK for Go. Se recomienda usar la versión más reciente.

Descripción

La exportación paralela primero llama a ComputeSplits para crear una sesión de escaneo y obtener la concurrencia recomendada. Después, cada trabajador concurrente invoca ParallelScan con un CurrentParallelID único y usa NextToken para leer su división de forma continua. El escaneo paralelo no garantiza el orden del conjunto de resultados global, por lo que es adecuado para exportaciones grandes que no dependen del orden de los resultados. También puede establecer MaxParallel en 1 y CurrentParallelID en 0 para ejecutar un escaneo con un solo trabajador. El código es más sencillo y el rendimiento suele ser superior al de Search, aunque inferior al de un escaneo con varios trabajadores.

Importante

El escaneo paralelo no admite ordenamiento ni agregaciones. Use Search si necesita ordenar resultados, realizar agregaciones o devolver resultados de búsqueda a los usuarios finales.
Los trabajadores que usan el mismo SessionId establecen una instantánea de datos en la primera llamada a ParallelScan. Los datos insertados o actualizados durante la tarea no se incluyen en la instantánea. Es posible omitir SessionId, pero cambios como el balanceo de carga del lado del servidor pueden provocar un pequeño número de filas duplicadas. Se recomienda llamar primero a ComputeSplits e incluir el SessionId devuelto en las solicitudes posteriores.
Se pueden ejecutar hasta 10 tareas de escaneo paralelo simultáneamente en el mismo índice de búsqueda. Para consultar otros límites, consulte Search index limits.

A continuación, el ejemplo inicia varias goroutines según la concurrencia recomendada que devuelve ComputeSplits, asigna un CurrentParallelID único a cada trabajador y lee todas las divisiones.

splits, err := client.ComputeSplits(
    (&tablestore.ComputeSplitsRequest{}).
        SetTableName("example_table").
        SetSearchIndexSplitsOptions(tablestore.SearchIndexSplitsOptions{
            IndexName: "example_index",
        }),
)
if err != nil {
    log.Fatal(err)
}

var waitGroup sync.WaitGroup
var mutex sync.Mutex
totalRows := 0
errors := make(chan error, splits.SplitsSize)

waitGroup.Add(int(splits.SplitsSize))
for workerID := int32(0); workerID < splits.SplitsSize; workerID++ {
    currentWorkerID := workerID
    go func() {
        defer waitGroup.Done()

        scanQuery := search.NewScanQuery().
            SetQuery(&search.MatchAllQuery{}).
            SetLimit(1000).
            SetMaxParallel(splits.SplitsSize).
            SetCurrentParallelID(currentWorkerID)
        request := (&tablestore.ParallelScanRequest{}).
            SetTableName("example_table").
            SetIndexName("example_index").
            SetScanQuery(scanQuery).
            SetSessionId(splits.SessionId).
            SetColumnsToGet(&tablestore.ColumnsToGet{
                ReturnAllFromIndex: true,
            })

        for {
            response, err := client.ParallelScan(request)
            if err != nil {
                errors <- err
                return
            }

            // Process response.Rows here.
            mutex.Lock()
            totalRows += len(response.Rows)
            mutex.Unlock()

            if len(response.NextToken) == 0 {
                return
            }
            request.SetScanQuery(scanQuery.SetToken(response.NextToken))
        }
    }()
}

waitGroup.Wait()
close(errors)
for err := range errors {
    log.Fatal(err)
}

fmt.Println(totalRows)

Parámetros

Crear una sesión de escaneo

Nombre	Tipo	Descripción
TableName (obligatorio)	string	El nombre de la tabla de datos.
IndexName (obligatorio)	string	El nombre del índice de búsqueda.

Solicitud de escaneo paralelo

Nombre	Tipo	Descripción
TableName (obligatorio)	string	El nombre de la tabla de datos.
IndexName (obligatorio)	string	El nombre del índice de búsqueda.
ScanQuery (obligatorio)	search.ScanQuery	La condición de escaneo y la configuración de concurrencia.
SessionId (opcional)	[]byte	El ID de sesión que devuelve ComputeSplits. Se recomienda especificar este parámetro para usar la misma instantánea de datos durante todo el escaneo.
ColumnsToGet (opcional)	*tablestore.ColumnsToGet	La configuración de columnas a devolver. Si se omite, solo se devuelven las columnas de clave principal. ParallelScan no admite ReturnAll.
TimeoutMs (opcional)	*int32	El tiempo de espera de la solicitud en milisegundos.

Configuración de escaneo

Nombre	Tipo	Descripción
Query (obligatorio)	search.Query	La condición de escaneo. Se admiten los tipos de consulta no vectoriales compatibles con Search.
Limit (opcional)	int32	El número máximo de filas devueltas por solicitud. Valor predeterminado: 2,000. Se recomienda mantener el valor predeterminado. El servidor permite un valor de hasta 10,000, pero un valor mayor incrementa la latencia de la solicitud y el uso de recursos.
MaxParallel (opcional)	int32	El número total de trabajadores. Valor predeterminado: 1. El valor no puede exceder el SplitsSize que devuelve ComputeSplits.
CurrentParallelID (opcional)	int32	El ID del trabajador actual. Este parámetro es obligatorio cuando MaxParallel es mayor que 1. Los ID de los trabajadores deben ser únicos y estar dentro del rango [0, MaxParallel).
Token (opcional)	[]byte	El valor NextToken que devolvió la respuesta anterior.
AliveTime (opcional)	int32	El intervalo máximo entre dos solicitudes de paginación, en segundos. Valores válidos: de 1 a 600. Valor predeterminado: 60. Se recomienda mantener el valor predeterminado. Cada solicitud de datos exitosa renueva el periodo de validez.

Configuración de columnas a devolver

Nombre	Tipo	Descripción
Columns (opcional)	[]string	Los campos del índice de búsqueda que se devolverán. No se puede devolver un campo que solo existe en la tabla de datos y no está incluido en el índice de búsqueda.
ReturnAllFromIndex (opcional)	bool	Especifica si se deben devolver todos los campos del índice de búsqueda. Valor predeterminado: false. Si este parámetro es true, no es necesario especificar Columns.
ReturnAll (opcional)	bool	El escaneo paralelo no admite este parámetro. No lo establezca en true.

Nota

Los cambios de esquema que alternan índices, la conmutación por error del servidor o el balanceo de carga pueden invalidar una sesión antes de tiempo y devolver OTSSessionExpired. Los errores de red del cliente también pueden interrumpir un escaneo. Si ocurre este tipo de error, descarte los resultados incompletos, llame nuevamente a ComputeSplits y reinicie toda la tarea de escaneo desde el principio.

Respuesta

Información de divisiones

Nombre	Tipo	Descripción
SessionId	[]byte	El ID de sesión de la tarea que se usa para escanear la misma instantánea de datos.
SplitsSize	int32	La concurrencia máxima admitida para el índice de búsqueda.

Resultado del escaneo

Nombre	Tipo	Descripción
Rows	[]*tablestore.Row	Las filas que devuelve el escaneo actual.
NextToken	[]byte	El token para la siguiente página. Continúe escaneando la división actual si el valor no está vacío.