Go SDK并发导出数据_表格存储(Tablestore)-阿里云帮助中心

当使用场景中不关心整个结果集的顺序时，您可以使用并发导出数据功能以更快的速度将命中的数据全部返回。

前提条件

已初始化Client。具体操作，请参见初始化OTSClient。
已创建数据表并写入数据。具体操作，请参见创建数据表和写入数据。
已在数据表上创建多元索引。具体操作，请参见创建多元索引。

参数

参数		说明
TableName		数据表名称。
IndexName		多元索引名称。
ScanQuery	Query	多元索引的查询语句。支持精确查询、模糊查询、范围查询、地理位置查询、嵌套查询等，功能和Search接口一致。
	Limit	扫描数据时一次能返回的数据行数。
	MaxParallel	最大并发数。请求支持的最大并发数由用户数据量决定。数据量越大，支持的并发数越多，每次任务前可以通过ComputeSplits API进行获取。
	CurrentParallelID	当前并发ID。取值范围为[0, MaxParallel)。
	Token	用于翻页功能。ParallelScan请求结果中有下一次进行翻页的Token，使用该Token可以接着上一次的结果继续读取数据。
	AliveTime	ParallelScan的当前任务有效时间，也是Token的有效时间。默认值为60，建议使用默认值，单位为秒。如果在有效时间内没有发起下一次请求，则不能继续读取数据。持续发起请求会刷新Token有效时间。说明动态修改schema中的切换索引、服务端单台机器故障、服务端负载均衡等均会导致Session提前过期，此时需要重新创建Session。
ColumnsToGet		指定分组结果中需要返回的列名，可以通过将列名加入Columns来实现。如果需要返回多元索引中的所有列，则可以使用更简洁的ReturnAllFromIndex实现。重要此处不能使用ReturnAll。
SessionId		本次并发扫描数据任务的SessionId。您可以通过ComputeSplits API创建Session，同时获得本次任务支持的最大并发数。

示例

请根据实际选择单并发扫描数据和多线程并发扫描数据。

单并发扫描数据

相对于多并发扫描数据，单并发扫描数据的代码更简单，单并发代码无需关心currentParallelId和maxParallel参数。单并发使用方式的整体吞吐比Search接口方式高，但是比多线程多并发使用方式的吞吐低。

func computeSplits(client *tablestore.TableStoreClient, tableName string, indexName string) (*tablestore.ComputeSplitsResponse, error) {
	req := &tablestore.ComputeSplitsRequest{}
	req.
		SetTableName(tableName).
		SetSearchIndexSplitsOptions(tablestore.SearchIndexSplitsOptions{IndexName: indexName})
	res, err := client.ComputeSplits(req)
	if err != nil {
		return nil, err
	}
	return res, nil
}

/**
 * ParallelScan单并发扫描数据。
 */
func ParallelScanSingleConcurrency(client *tablestore.TableStoreClient, tableName string, indexName string) {
        computeSplitsResp, err := computeSplits(client, tableName, indexName)
        if err != nil {
                fmt.Printf("%#v", err)
                return
        }

        query := search.NewScanQuery().SetQuery(&search.MatchAllQuery{}).SetLimit(2)

        req := &tablestore.ParallelScanRequest{}
        req.SetTableName(tableName).
                SetIndexName(indexName).
                SetColumnsToGet(&tablestore.ColumnsToGet{ReturnAllFromIndex: false}).
                SetScanQuery(query).
                SetSessionId(computeSplitsResp.SessionId)

        res, err := client.ParallelScan(req)
        if err != nil {
                fmt.Printf("%#v", err)
                return
        }

        total := len(res.Rows)
        for res.NextToken != nil {
                req.SetScanQuery(query.SetToken(res.NextToken))
                res, err = client.ParallelScan(req)
                if err != nil {
                        fmt.Printf("%#v", err)
                        return
                }

                total += len(res.Rows) //process rows each loop
        }
        fmt.Println("total: ", total)
}

多线程并发扫描数据

func computeSplits(client *tablestore.TableStoreClient, tableName string, indexName string) (*tablestore.ComputeSplitsResponse, error) {
	req := &tablestore.ComputeSplitsRequest{}
	req.
		SetTableName(tableName).
		SetSearchIndexSplitsOptions(tablestore.SearchIndexSplitsOptions{IndexName: indexName})
	res, err := client.ComputeSplits(req)
	if err != nil {
		return nil, err
	}
	return res, nil
}

/**
 * ParallelScan多并发扫描数据。
 */
func ParallelScanMultiConcurrency(client *tablestore.TableStoreClient, tableName string, indexName string) {
        computeSplitsResp, err := computeSplits(client, tableName, indexName)
        if err != nil {
                fmt.Printf("%#v", err)
                return
        }

        var wg sync.WaitGroup
        wg.Add(int(computeSplitsResp.SplitsSize))

        for i := int32(0); i < computeSplitsResp.SplitsSize; i++ {
                current := i
                go func() {
                        defer wg.Done()
                        query := search.NewScanQuery().
                                SetQuery(&search.MatchAllQuery{}).
                                SetCurrentParallelID(current).
                                SetMaxParallel(computeSplitsResp.SplitsSize).
                                SetLimit(2)

                        req := &tablestore.ParallelScanRequest{}
                        req.SetTableName(tableName).
                                SetIndexName(indexName).
                                SetColumnsToGet(&tablestore.ColumnsToGet{ReturnAllFromIndex: false}).
                                SetScanQuery(query).
                                SetSessionId(computeSplitsResp.SessionId)

                        res, err := client.ParallelScan(req)
                        if err != nil {
                                fmt.Printf("%#v", err)
                                return
                        }

                        total := len(res.Rows)
                        for res.NextToken != nil {
                                req.SetScanQuery(query.SetToken(res.NextToken))
                                res, err = client.ParallelScan(req)
                                if err != nil {
                                        fmt.Printf("%#v", err)
                                        return
                                }

                                total += len(res.Rows) //process rows each loop
                        }
                        fmt.Println("total: ", total)
                }()
        }
        wg.Wait()
}

常见问题

使用多元索引Search接口查不到数据

前提条件

参数

示例

单并发扫描数据

多线程并发扫描数据

常见问题

相关文档