Tablestore:Pemindaian paralel

Prasyarat

Sebelum memulai, lakukan persiapan berikut:

• Instal Tablestore Go SDK dan inisialisasi client.

• Pemindaian paralel memerlukan Tablestore SDK untuk Go versi 1.6.0 atau lebih baru. Disarankan untuk menggunakan versi terbaru.

Deskripsi

Ekspor paralel dimulai dengan memanggil ComputeSplits untuk membuat sesi pemindaian dan mendapatkan tingkat konkurensi yang direkomendasikan. Setiap worker konkuren kemudian memanggil ParallelScan dengan CurrentParallelID unik dan menggunakan NextToken untuk membaca bagian datanya secara berkelanjutan. Pemindaian paralel tidak menjamin urutan set hasil secara keseluruhan, sehingga cocok untuk ekspor skala besar yang tidak bergantung pada urutan hasil. Anda juga dapat mengatur MaxParallel ke 1 dan CurrentParallelID ke 0 untuk pemindaian satu worker. Pendekatan ini menghasilkan kode yang lebih sederhana, dengan throughput biasanya lebih tinggi daripada Search namun lebih rendah dibandingkan pemindaian multi-worker.

Contoh berikut memulai beberapa goroutine berdasarkan tingkat konkurensi yang direkomendasikan oleh ComputeSplits, memberikan CurrentParallelID unik untuk setiap worker, dan membaca semua bagian data.

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
            }

            // Proses response.Rows di sini.
            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)

Parameter

Respons