Use Tablestore SDK for Go to perform a parallel scan - Tablestore: Central de Documentação do Alibaba Cloud

Se a ordem dos resultados da consulta não for relevante, use o recurso de varredura paralela para obter os dados com eficiência.

Pré-requisitos

Instância OTSClient inicializada. Para mais informações, consulte Inicializar uma instância OTSClient.
Tabela de dados criada e preenchida. Para mais informações, consulte Criar uma tabela de dados e Gravar dados.
Índice de pesquisa criado para a tabela de dados. Para mais informações, consulte Criar índices de pesquisa.

Parâmetros

Parâmetro		Descrição
TableName		Nome da tabela de dados.
IndexName		Nome do índice de pesquisa.
ScanQuery	Query	Instrução de consulta do índice de pesquisa. Aceita tipos como term query, fuzzy query, range query, geo query e nested query, assim como a operação Search.
	Limit	Número máximo de linhas retornadas por chamada ParallelScan.
	MaxParallel	Número máximo de tarefas de varredura paralela por requisição. Esse valor varia conforme o volume de dados: volumes maiores exigem mais tarefas paralelas. Chame a operação ComputeSplits para consultar o limite máximo de tarefas paralelas por requisição.
	CurrentParallelID	ID da tarefa de varredura paralela na requisição. Valores válidos: [0, MaxParallel).
	Token	Token usado para paginar resultados da consulta. A resposta da requisição ParallelScan inclui o token da próxima página para recuperação dos dados subsequentes.
	AliveTime	Período de validade da tarefa de varredura paralela atual e do token. Valor padrão: 60 segundos. Mantenha o valor padrão. Se a próxima requisição não ocorrer dentro desse período, a consulta de dados adicionais falhará. O envio de cada requisição renova a validade do token. Nota As sessões expiram antecipadamente se houver modificação dinâmica de índices de troca nos esquemas, falha em um único servidor ou balanceamento de carga do servidor. Nesse caso, recrie as sessões.
ColumnsToGet		Nome da coluna a retornar no resultado do agrupamento. Defina o parâmetro Columns com o nome da coluna desejada. Para retornar todas as colunas do índice de pesquisa, especifique o parâmetro ReturnAllFromIndex. Importante O parâmetro ReturnAll não é suportado.
SessionId		ID da sessão da tarefa de varredura paralela. Chame a operação ComputeSplits para criar uma sessão e consultar o número máximo de tarefas de varredura paralela que a requisição suporta.

Exemplos

Conforme a necessidade do seu negócio, execute a varredura dos dados com uma única thread ou com múltiplas threads simultaneamente.

Varredura de dados com thread única

O código de requisição com thread única para varredura paralela é mais simples que o de múltiplas threads. Os parâmetros currentParallelId e maxParallel são opcionais nesse cenário. A requisição ParallelScan com thread única oferece throughput superior ao da requisição Search, mas inferior ao da ParallelScan com múltiplas threads.

func computeSplits(client *tablestore.TableStoreClient, tableName string, indexName string) (*tablestore.ComputeSplitsResponse, error) {
	req := &tablestore.ComputeSplitsRequest{}
	req.
		SetTableName(tableName).
		SetSearchIndexSplitsOptions(tablestore.SearchIndexSplitsOptions{IndexName: indexName})
	res, err := client.ComputeSplits(req)
	if err != nil {
		return nil, err
	}
	return res, nil
}

/**
 * Perform a parallel scan to scan data by using a single thread. 
 */
func ParallelScanSingleConcurrency(client *tablestore.TableStoreClient, tableName string, indexName string) {
        computeSplitsResp, err := computeSplits(client, tableName, indexName)
        if err != nil {
                fmt.Printf("%#v", err)
                return
        }

        query := search.NewScanQuery().SetQuery(&search.MatchAllQuery{}).SetLimit(2)

        req := &tablestore.ParallelScanRequest{}
        req.SetTableName(tableName).
                SetIndexName(indexName).
                SetColumnsToGet(&tablestore.ColumnsToGet{ReturnAllFromIndex: false}).
                SetScanQuery(query).
                SetSessionId(computeSplitsResp.SessionId)

        res, err := client.ParallelScan(req)
        if err != nil {
                fmt.Printf("%#v", err)
                return
        }

        total := len(res.Rows)
        for res.NextToken != nil {
                req.SetScanQuery(query.SetToken(res.NextToken))
                res, err = client.ParallelScan(req)
                if err != nil {
                        fmt.Printf("%#v", err)
                        return
                }

                total += len(res.Rows) //process rows each loop
        }
        fmt.Println("total: ", total)
}

Varredura de dados com múltiplas threads

func computeSplits(client *tablestore.TableStoreClient, tableName string, indexName string) (*tablestore.ComputeSplitsResponse, error) {
	req := &tablestore.ComputeSplitsRequest{}
	req.
		SetTableName(tableName).
		SetSearchIndexSplitsOptions(tablestore.SearchIndexSplitsOptions{IndexName: indexName})
	res, err := client.ComputeSplits(req)
	if err != nil {
		return nil, err
	}
	return res, nil
}

/**
 * Perform a parallel scan to scan data by using multiple threads. 
 */
func ParallelScanMultiConcurrency(client *tablestore.TableStoreClient, tableName string, indexName string) {
        computeSplitsResp, err := computeSplits(client, tableName, indexName)
        if err != nil {
                fmt.Printf("%#v", err)
                return
        }

        var wg sync.WaitGroup
        wg.Add(int(computeSplitsResp.SplitsSize))

        for i := int32(0); i < computeSplitsResp.SplitsSize; i++ {
                current := i
                go func() {
                        defer wg.Done()
                        query := search.NewScanQuery().
                                SetQuery(&search.MatchAllQuery{}).
                                SetCurrentParallelID(current).
                                SetMaxParallel(computeSplitsResp.SplitsSize).
                                SetLimit(2)

                        req := &tablestore.ParallelScanRequest{}
                        req.SetTableName(tableName).
                                SetIndexName(indexName).
                                SetColumnsToGet(&tablestore.ColumnsToGet{ReturnAllFromIndex: false}).
                                SetScanQuery(query).
                                SetSessionId(computeSplitsResp.SessionId)

                        res, err := client.ParallelScan(req)
                        if err != nil {
                                fmt.Printf("%#v", err)
                                return
                        }

                        total := len(res.Rows)
                        for res.NextToken != nil {
                                req.SetScanQuery(query.SetToken(res.NextToken))
                                res, err = client.ParallelScan(req)
                                if err != nil {
                                        fmt.Printf("%#v", err)
                                        return
                                }

                                total += len(res.Rows) //process rows each loop
                        }
                        fmt.Println("total: ", total)
                }()
        }
        wg.Wait()
}

Perguntas frequentes

O que fazer quando nenhum dado é encontrado ao chamar a operação Search para índices de pesquisa?

Referências

A consulta de dados com índice de pesquisa oferece os seguintes métodos: term query, terms query, match all query, match query, match phrase query, prefix query, range query, wildcard query, geo query, Boolean query, KNN vector query, nested query e exists query. Use esses métodos para consultar dados sob múltiplas dimensões conforme a necessidade do seu negócio.

Ordene ou pagine as linhas que atendem aos critérios de consulta pelos recursos de ordenação e paginação. Para mais informações, consulte Ordenação e paginação.

O recurso de colapso (distinct) agrupa o conjunto de resultados com base em uma coluna específica. Assim, dados do tipo especificado aparecem apenas uma vez nos resultados da consulta. Para mais informações, consulte Colapso (deduplicação).
Para analisar dados em uma tabela, use o recurso de agregação da operação Search ou execute instruções SQL. É possível obter valores mínimos e máximos, somas e contagem total de linhas, por exemplo. Para mais informações, consulte Agregação e Consulta SQL.
Para obter todas as linhas correspondentes aos critérios de consulta sem ordenação, chame as operações ParallelScan e ComputeSplits para usar o recurso de varredura paralela. Para mais informações, consulte Executar uma varredura paralela.