Tablestore：並發匯出資料

前提條件

開始前，完成以下準備工作：

• 安裝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)

參數說明

傳回值