Tablestore:Parallel scan

Prerequisites

Before you begin, complete the following preparations:

• Install the Tablestore Go SDK and initialize a client.

• Parallel scan requires Tablestore SDK for Go 1.6.0 or later. We recommend that you use the latest version.

Description

Parallel export first calls ComputeSplits to create a scan session and obtain the recommended concurrency. Then each concurrent worker calls ParallelScan with a unique CurrentParallelID and uses NextToken to read its split continuously. Parallel scan does not guarantee the order of the overall result set and is suitable for large exports that do not depend on result order. You can also set MaxParallel to 1 and CurrentParallelID to 0 for a single-worker scan. The code is simpler, and throughput is typically higher than Search but lower than a multi-worker scan.

The following example starts multiple goroutines based on the recommended concurrency returned by ComputeSplits, assigns a unique CurrentParallelID to each worker, and reads all splits.

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)

Parameters

Response