Table Store Python SDK 並發匯出資料 - 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))