并发导出数据

前提条件

开始前，完成以下准备工作：

• 安装Tablestore Go SDK并初始化客户端。

• 并发导出数据需要 Go SDK 1.6.0 及以上版本，建议使用最新版本。

功能说明

并发导出先调用 ComputeSplits 创建扫描会话并获取建议并发数，再为每个并发任务调用 ParallelScan。每个并发任务使用独立的 CurrentParallelID，并通过 NextToken 连续读取该分片。并发扫描不保证整个结果集的顺序，适合不依赖返回顺序的大规模导出。也可以将 MaxParallel 设置为 1、CurrentParallelID 设置为 0 进行单并发扫描；代码更简单，吞吐通常高于 Search，但低于多并发扫描。

以下示例根据 ComputeSplits 返回的建议并发数启动多个 goroutine，并为每个并发任务设置不同的 CurrentParallelID，直至读取完所有分片。

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)

参数说明

返回值