Tablestore:並列スキャン

前提条件

開始する前に、次の準備を完了してください。

• Tablestore Go SDK をインストールし、クライアントを初期化します。

• 並列スキャンには Tablestore SDK for Go 1.6.0 以降が必要です。最新バージョンの使用を推奨します。

説明

並列エクスポートでは、最初に ComputeSplits を呼び出してスキャンセッションを作成し、推奨される同時実行数を取得します。次に、同時実行される各ワーカーが、一意の CurrentParallelID を指定して ParallelScan を呼び出し、NextToken を使用して自身のスプリットを継続的に読み取ります。並列スキャンは結果セット全体の順序を保証しないため、結果の順序に依存しない大規模なエクスポートに適しています。単一ワーカーでスキャンする場合は、MaxParallel を 1、CurrentParallelID を 0 に設定することもできます。この場合、コードはよりシンプルになり、スループットは通常、検索よりも高い一方で、複数ワーカーによるスキャンよりは低くなります。

次の例では、ComputeSplits から返される推奨同時実行数に基づいて複数のゴルーチンを開始し、各ワーカーに一意の 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
            }

            // ここで response.Rows を処理します。
            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)

パラメータ

レスポンス