Parallel scan - Tablestore SDK for Python - Tablestore - Alibaba Cloud Documentation Center

Prerequisites

Install the Tablestore SDK for Python and initialize a client.

Description

Parallel scan exports all rows that match a query in a search index. Results are not globally ordered, and sorting and aggregation are not supported. To sort or aggregate results or return search results to end users, use the Search API. A single worker is simpler to configure, whereas multiple workers read multiple splits concurrently and typically provide higher throughput.

The workflow is: Call compute_splits to obtain the maximum concurrency splits_size and session identifier session_id. Configure each worker with the same query, session_id, and max_parallel, and a unique current_parallel_id. Each worker calls parallel_scan and uses next_token to read subsequent pages. Finally, wait for all workers and merge their unordered results.

Important

Workers under the same session_id establish a data snapshot when the first scan starts. Dynamic schema updates, failover, or load balancing can expire a session early and return OTSSessionExpired; network failures can also interrupt a scan. In these cases, discard incomplete results, call compute_splits again, and restart all workers from the beginning. Up to 10 parallel scan jobs can run concurrently on one search index.

The following example computes splits and then scans all data by using one worker.

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))

Parameters

Compute splits

compute_splits(table_name, index_name) contains the following parameters.

Name	Type	Description
table_name (required)	`str`	The name of the data table.
index_name (required)	`str`	The name of the search index.

Scan request

parallel_scan contains the following parameters.

Name	Type	Description
table_name (required)	`str`	The name of the data table.
index_name (required)	`str`	The name of the search index.
scan_query (required)	`ScanQuery`	The scan condition, pagination, and concurrency configurations.
session_id (required)	`bytes`	The session identifier returned by `compute_splits`, used to maintain the same data snapshot.
columns_to_get (optional)	`ColumnsToGet`	The return column configuration. If omitted, only primary key columns are returned.
timeout_s (optional)	`int`	The request timeout in seconds.

Scan configuration

scan_query is of the ScanQuery type and contains the following parameters.

Name	Type	Description
query (required)	`Query`	The query condition that defines the scan scope. Use `MatchAllQuery` to scan all rows.
limit (required)	`int`	The maximum number of rows per request. We recommend the default value `2000`. The server allows up to `10000`, but using the maximum is not recommended.
next_token (required)	`bytes`	The pagination token. Set it to `None` in the first request and use `next_token` from the previous response in subsequent requests.
current_parallel_id (required)	`int`	The current worker ID. Valid values: `[0, max_parallel)`. Each worker must use a unique value.
max_parallel (required)	`int`	The job concurrency, which cannot exceed `splits_size`.
alive_time (optional)	`int`	The maximum validity period between two page requests, in seconds. Valid values: 1 to 600. Default value: `60`. The period is refreshed after a successful response.

Return columns

columns_to_get is of the ColumnsToGet type and contains the following parameters.

Name	Type	Description
column_names (optional)	`list[str]`	The names of search index fields to return. Specify this parameter only when `return_type` is `SPECIFIED`.
return_type (optional)	`ColumnReturnType`	The return column mode. Parallel scan supports `NONE`, `SPECIFIED`, and `ALL_FROM_INDEX`, but not `ALL`.

Response

Split information

compute_splits returns split information.

Field	Type	Description
session_id	`bytes`	The job session identifier.
splits_size	`int`	The maximum concurrency supported by the search index.

Scan result

parallel_scan returns a scan result.

Field	Type	Description
rows	`list[Row]`	The rows returned by the request.
next_token	`bytes`	The token for the next page. An empty value indicates that the current worker has finished.

Tuple-compatible responses

Parallel scan is supported starting from Tablestore SDK for Python 5.2.0, which returns response objects. In version 5.2.1 and later, you can call ComputeSplitsResponse.v1_response() and ParallelScanResponse.v1_response() to obtain tuples. For new code, access response object attributes directly.

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

Examples

Scan with multiple workers

The following example creates workers based on splits_size. The thread pool does not exceed the number of client CPU cores, and all workers share the session and maximum concurrency.

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))