Sharding and concurrency#

Two independent dials#

The flag names do not make this obvious, so it is worth stating first:

--max-concurrency: how many requests are in flight at once, within a shard
--num-threads: how many shards the input is split into

They multiply. Four shards at concurrency 8 is up to 32 simultaneous requests against your endpoint. Tuning one without noticing the other is the usual reason a swarm gets overwhelmed.

Bounded concurrency#

--max-concurrency is enforced with an asyncio.Semaphore and a matching worker count, so it is a hard ceiling on in-flight requests, not a target.

Raise it until the endpoint is saturated and no further. Past that point extra concurrency adds queueing latency and makes timeouts more likely without adding throughput — the replicas are already busy.

Retries and timeouts#

--retries sets the maximum attempts per item, applied with tenacity’s exponential backoff: multiplier 1, minimum 4 seconds, capped at 10.

--timeout is the per-request timeout in seconds. It interacts with concurrency: if concurrency is high enough that requests queue at the endpoint, a timeout tuned for an idle endpoint will start firing under load. Symptoms of that are covered in Monitoring and troubleshooting.

Sharding the input#

--num-threads splits the input into that many shards, run concurrently, each with its own checkpoint directory. That per-shard isolation is what makes concurrent flushing safe.

Important

Sharded execution requires checkpointing. Combining --num-threads > 1 with --no-checkpointing raises:

Sharded execution requires checkpointing to be enabled.

Shard results are assembled from checkpoint state, so there is nothing to assemble without it.

Choosing a shard mode#

--shard-mode takes exactly id or index; anything else raises a ValueError.

id (default): assigns each row by a stable hash of its id — hash_array(ids) % nshards. A given row lands in the same shard regardless of where it sits in the input, or how long the input is.
index: splits row order into contiguous blocks with np.array_split. This is the legacy behaviour.

The consequence matters for resume. With index, inserting a row near the start shifts everything after it into different shards, and previously completed work becomes invisible to the shard now responsible for it. With id, row identity survives reordering and length changes.

Prefer id unless you specifically need contiguous blocks — and pair it with a real --id-column, since a generated index is not stable across scans.

Resume across a changed shard count#

Even with id, changing --num-threads changes nshards and therefore the modulo, so rows move between shards. A shard will not find another shard’s checkpoint.

Two ways through:

keep --num-threads fixed across resumed runs, which is the simple answer
pass --global-resume, which filters the input against the union of done ids across all shards rather than each shard’s own

The same applies to --limit, since it changes which rows are present. See Checkpointing and resuming.

Output layout#

Checkpoint stores are per shard, derived by inserting the shard index: run.parquet becomes run_shard0.parquet, run_shard1.parquet and so on.

For the final output, a single-file --output receives all shards merged. With a directory output plus --shard-output on the Polars runner, one Parquet file per shard is written instead, named zero-padded by shard count — data-00.parquet, data-01.parquet — so lexical order matches shard order. Checkpoint outputs are the source of truth for that write.

Directory outputs are the better choice for large results: no merge step, and downstream readers can consume the shards in parallel.

Ray#

Ray takes a different execution path and does not use --num-threads sharding — distribution is Ray’s own concern. It requires --native-backend and a --ray-address. See Choosing a data backend.

A starting point#

For a first run on an unfamiliar model and cluster:

--max-concurrency 8 --num-threads 1 --retries 3 --timeout 600 --limit 100

Confirm correctness on 100 rows, then raise --max-concurrency until throughput stops improving, then add shards with --num-threads if the driver process itself becomes the bottleneck rather than the endpoint.