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

前提条件

功能说明

并发导出数据用于全量扫描多元索引中满足查询条件的数据。结果不保证全局顺序，也不支持排序和统计聚合；如果需要排序、聚合或面向最终用户返回检索结果，请使用 Search 接口。单并发配置简单，多并发可同时读取多个分片，通常具有更高吞吐量。

完整流程为：调用 compute_splits 获取最大并发度 splits_size 和会话标识 session_id；为每个并发任务设置相同的查询条件、session_id 和 max_parallel，并使用不同的 current_parallel_id；各任务调用 parallel_scan，使用 next_token 读取自身后续分页；最后等待全部任务完成并合并无序结果。

重要

同一 session_id 下的任务在首次扫描时确定数据快照。会话可能因动态修改 Schema、故障转移或负载均衡提前失效并返回 OTSSessionExpired；网络异常也可能中断任务。发生异常时，丢弃当前不完整结果，重新调用 compute_splits 并从头启动全部任务。同一多元索引最多同时运行 10 个并发扫描任务。

以下示例先计算分片，再以单并发方式扫描全部数据。

splits = client.compute_splits("example_table", "example_index")
next_token = None
rows = []

while True:
    scan_query = ScanQuery(
        MatchAllQuery(),
        limit=2000,
        next_token=next_token,
        current_parallel_id=0,
        max_parallel=1,
        alive_time=60,
    )
    response = client.parallel_scan(
        "example_table",
        "example_index",
        scan_query,
        splits.session_id,
        ColumnsToGet(return_type=ColumnReturnType.ALL_FROM_INDEX),
    )
    rows.extend(response.rows)
    next_token = response.next_token
    if not next_token:
        break

print(len(rows))

参数说明

计算分片

compute_splits(table_name, index_name) 包含以下参数。

名称	类型	说明
table_name（必选）	`str`	数据表名称。
index_name（必选）	`str`	多元索引名称。

扫描请求

parallel_scan 包含以下参数。

名称	类型	说明
table_name（必选）	`str`	数据表名称。
index_name（必选）	`str`	多元索引名称。
scan_query（必选）	`ScanQuery`	扫描条件、分页和并发配置。
session_id（必选）	`bytes`	`compute_splits` 返回的会话标识，用于保持同一数据快照。
columns_to_get（可选）	`ColumnsToGet`	返回列配置。未设置时只返回主键列。
timeout_s（可选）	`int`	请求级超时时间，单位为秒。

扫描配置

scan_query 的类型为 ScanQuery，包含以下参数。

名称	类型	说明
query（必选）	`Query`	扫描范围对应的查询条件。扫描全部数据时设置为 `MatchAllQuery`。
limit（必选）	`int`	单次请求最大返回行数，默认建议使用 `2000`。服务端允许最大值 `10000`，但不建议设置为该上限。
next_token（必选）	`bytes`	分页凭证。首次请求设置为 `None`，后续请求使用上一次响应的 `next_token`。
current_parallel_id（必选）	`int`	当前并发任务 ID，取值范围为 `[0, max_parallel)`，各任务不能重复。
max_parallel（必选）	`int`	任务并发度，不能大于 `splits_size`。
alive_time（可选）	`int`	两次分页请求之间的最大有效时间，单位为秒，取值范围为 1～600，默认值为 `60`。成功获取数据后会刷新。

返回列

columns_to_get 的类型为 ColumnsToGet，包含以下参数。

名称	类型	说明
column_names（可选）	`list[str]`	要返回的多元索引字段名称。仅 `return_type` 为 `SPECIFIED` 时设置。
return_type（可选）	`ColumnReturnType`	返回列模式。并发扫描支持 `NONE`、`SPECIFIED` 和 `ALL_FROM_INDEX`，不支持 `ALL`。

返回值

分片信息

compute_splits 返回分片信息。

字段	类型	说明
session_id	`bytes`	任务会话标识。
splits_size	`int`	多元索引支持的最大并发度。

扫描结果

parallel_scan 返回扫描结果。

字段	类型	说明
rows	`list[Row]`	本次请求返回的数据行。
next_token	`bytes`	下一页凭证。值为空时当前并发任务已读取完毕。

兼容 Tuple 返回格式

并发导出功能从 Tablestore Python SDK 5.2.0 开始支持，5.2.0 版本返回响应对象。5.2.1 及以上版本可分别调用 ComputeSplitsResponse.v1_response() 和 ParallelScanResponse.v1_response() 获取 Tuple。新代码建议直接访问响应对象的属性。

session_id, splits_size = splits.v1_response()
rows, next_token = response.v1_response()

场景示例

多并发扫描

以下示例按 splits_size 创建多个任务。线程池大小不超过客户端 CPU 核数，所有任务共享会话和最大并发度。

from concurrent.futures import ThreadPoolExecutor
import os


def scan_split(parallel_id, max_parallel, session_id):
    rows = []
    next_token = None
    while True:
        scan_query = ScanQuery(
            MatchAllQuery(),
            limit=2000,
            next_token=next_token,
            current_parallel_id=parallel_id,
            max_parallel=max_parallel,
            alive_time=60,
        )
        response = client.parallel_scan(
            "example_table",
            "example_index",
            scan_query,
            session_id,
            ColumnsToGet(return_type=ColumnReturnType.ALL_FROM_INDEX),
        )
        rows.extend(response.rows)
        next_token = response.next_token
        if not next_token:
            return rows


splits = client.compute_splits("example_table", "example_index")
worker_count = min(splits.splits_size, os.cpu_count() or 1)
with ThreadPoolExecutor(max_workers=worker_count) as executor:
    futures = [
        executor.submit(
            scan_split,
            parallel_id,
            splits.splits_size,
            splits.session_id,
        )
        for parallel_id in range(splits.splits_size)
    ]
    all_rows = [row for future in futures for row in future.result()]

print(len(all_rows))