Many tables¶

A stream reads one capture instance. For a database with many tracked tables, run one stream per table: start_many starts one to_delta stream per capture instance in the same Spark session, each with its own checkpoint, app_id and bronze table, all writing to one facts table (ADR 0027).

Start them together¶

A scheduled job that catches every table up, then advances each table's verdict:

from mssql_cdc import await_all, finalization, start_many

options = {
    "connectionString": "Server=sqlhost,1433;Database=sales;Encrypt=yes",
    "user": "cdc_reader",
    "password": "...",
}
queries = start_many(
    spark,
    options,
    ["dbo_orders", "dbo_order_items", "dbo_customers"],
    target="bronze.{ci}",
    app_id="{ci}-v1",
    checkpoint="/Volumes/main/ops/checkpoints/{ci}",
    facts_table="ops.ingestion_facts",
    trigger={"availableNow": True},
)
failed = await_all(queries)

for ci, query in queries.items():
    if ci not in failed:
        end = finalization.end_offset_from_progress(query.lastProgress)
        finalization.advance(spark, "ops.table_finalization", f"bronze.{ci}", end)
if failed:
    raise RuntimeError(f"CDC streams failed: {sorted(failed)}")  # fail the run, alert

target, app_id and checkpoint are templates: {ci} becomes the capture instance. Each must contain it, or start_many raises ValueError before starting anything. The other keyword arguments (facts_table, trigger, bootstrap, on_data_loss...) go to every to_delta, as in Options.

await_all waits for every query and returns the error of each one that failed, by capture instance. It raises nothing, so the job above raises itself: without that, a run with a failed table succeeds. is_data_loss(error) and is_schema_changed(error) pick out the failures that need a decision rather than a retry.

Options per table¶

Pass a mapping instead of a list to give a stream options of its own. They override the shared ones, matched ignoring case like every option:

queries = start_many(
    spark,
    options,
    {
        "dbo_orders": {"numPartitions": "8", "maxCommitsPerBatch": "500"},
        "dbo_order_items": {"numPartitions": "8"},
        "dbo_customers": {"numPartitions": "1"},
    },
    target="bronze.{ci}",
    app_id="{ci}-v1",
    checkpoint="/Volumes/main/ops/checkpoints/{ci}",
    facts_table="ops.ingestion_facts",
)

What each stream has¶

Each stream is exactly the one stream(spark, options).to_delta(...) starts for its table alone; start_many adds no state of its own.

• The facts table is created, or brought up to date, once before the first stream starts. Writers that commit to a new Delta table at the same time conflict; afterwards every stream only appends its own rows. The health query groups by target, so it covers every table as is.
• A shared facts table shows every table's facts to whoever reads it, but no source key values: a chunked snapshot keeps its chunk bounds in a plans table next to its bronze table (Bootstrap). One opened before 0.7.0 keeps them in its facts rows until it completes: until then, where keys are natural or personal identifiers, give the facts table the access policy of its bronze tables.
• Finalization stays per table: advance each target's verdict after its own query, as in the job above. A consumer that joins two tables waits until both are final for its period. Each stream is a prefix of the commit history of its own table, read at its own pace, so there is no single LSN across tables (Finalization).
• A new checkpoint template needs a new app_id template, as for one stream: bump the version in both ({ci}-v2, .../checkpoints-v2/{ci}) (Streaming).

Keep them running¶

For an always-on job, use a processingTime trigger and bound the run with a timeout:

from mssql_cdc import await_all, finalization, start_many, stop_all

tables = ["dbo_orders", "dbo_order_items", "dbo_customers"]
common = {
    "target": "bronze.{ci}",
    "app_id": "{ci}-v1",
    "checkpoint": "/Volumes/main/ops/checkpoints/{ci}",
    "facts_table": "ops.ingestion_facts",
    "trigger": {"processingTime": "1 minute"},
}
queries = start_many(spark, options, tables, **common)
trackers = [
    finalization.track(spark, q, "ops.table_finalization", f"bronze.{ci}")
    for ci, q in queries.items()
]
failed = await_all(queries, timeout=4 * 3600)  # every table runs for four hours
stop_all(queries)
for tracker in trackers:
    tracker.join()  # each table's last verdict is written, or its failure logged
if failed:
    raise RuntimeError(f"CDC streams failed: {sorted(failed)}")

A failed table stops alone while the others keep running; the run reports it when the timeout ends, and the next run (the job's schedule or retry) starts every table again from its checkpoint. In between, its facts stop arriving and the health query alerts on that target. await_all with a timeout returns the failures so far; the queries still running are not in it.

To stop everything as soon as one table fails instead, wait with Spark's own call:

try:
    spark.streams.awaitAnyTermination()  # returns, or raises its error, when one query stops
finally:
    stop_all(queries)

awaitAnyTermination keeps returning at once after the first query stopped, until spark.streams.resetTerminated().

Failure isolation¶

• A query that fails stops alone. Spark runs each query on its own thread with its own checkpoint; the others keep reading and writing (tests/test_fanout.py). Nothing in start_many stops them.
• Restart one table in the same session with the same arguments: it resumes from its checkpoint.

stopped = [ci for ci, q in queries.items() if not q.isActive]
queries.update(start_many(spark, options, stopped, **common))

Starting a table whose query still runs fails: Spark refuses two queries on one checkpoint. After DataLossError or SchemaChangedError, decide first (Data loss, Schema changes). - A start that raises stops the queries already started and raises the error: a wrong option, a missing permission on one table, a bootstrap that failed, or the on_data_loss="resnapshot" pre-flight refusing a second re-snapshot. These need a person; take that table out of the list (or into a job of its own) to run the others meanwhile. - Starts are sequential. to_delta runs a table's bootstrap or re-snapshot before it starts the query, so a large snapshot delays the tables after it in the list. For a first run over many large tables, put them last, or bootstrap them in a run of their own. The next table starts once the query before it has started its source, or stopped (below).

Sizing the cluster¶

The streams share the cluster, and their micro-batches run at the same time.

• Executor cores. A batch is read in up to numPartitions ranges, each a task on one core with its own connection to SQL Server. numPartitions=auto gives every stream all the session's cores (ADR 0011), but a range holds about 50,000 change rows or more (ADR 0015), so a table's steady small batches take one core each and only a large batch (a backlog, a bulk load) spreads over all of them. Cap it per table where those large batches would starve the other streams: 1 or 2 for small or quiet tables, more for the busy ones.
• Connections on SQL Server. At most one per running task, so no more than the executor cores, plus one per stream on the driver, which plans its batches.
• The driver plans every stream and runs every foreachBatch, where the sink writes bronze and the facts. Each stream also keeps, for the life of its query, a Python process of its own on the driver, where Spark runs the source's planning, and that process's connection to SQL Server. What that costs, and how many streams one driver holds, is below.
• Facts commits. Every trigger of every stream writes a facts commit. A processingTime trigger bounds them; compact the facts table (Monitoring).
• Scheduling. Spark schedules the jobs of all queries first in, first out by default, so a large batch of one table can hold the cores while the others wait. Cap it with maxCommitsPerBatch on that table, or give the queries fair scheduler pools.

How many streams per driver¶

What a running stream costs the driver, for the life of its query, measured with lab check t11 (ADR 0027):

Measured with 1, 5, 10 and 20 tables of the file-backed fake, a row a second each, a 5-second processingTime trigger, on a 16-vCPU host that other jobs kept busy. Memory is not what runs out first: 20 streams hold about 1 GB of Python processes. Two things are:

• Starting a stream. Spark gives each of those Python processes 10 seconds to start (import PySpark, PyArrow and this package) and connect back, and the limit cannot be configured. On a busy driver a start misses it: the query fails at once with "Python worker failed to connect back" (on Spark Connect, the foreachBatch process's miss makes start_many raise and stop the queries it started). start_many starts each stream once the one before it has started its source, so the processes do not start together; even so, 5 of 20 failed while 13 streams kept the measuring host busy, which other jobs also loaded (9 of 20 when started back to back; none of 20 back to back with the host otherwise idle). Start them again as under Failure isolation.
• The sink's work. Each trigger's foreachBatch commits bronze and the facts from the driver, which takes far longer than planning. When a trigger outlasts its interval, Spark logs "Current batch is falling behind": the driver, not SQL Server, sets the pace.

Start with up to about 10 always-on streams per driver, and watch for both errors. Past that:

• A longer trigger interval. A processingTime of minutes plans and commits less often per stream. The per-stream memory and threads stay as long as the query runs.
• Groups with availableNow. Each query stops once it has caught up, and its Python processes exit with it, so a driver holds only one group's at a time:

ids = {
    "target": "bronze.{ci}",
    "app_id": "{ci}-v1",
    "checkpoint": "/Volumes/main/ops/checkpoints/{ci}",
    "facts_table": "ops.ingestion_facts",
}
failed = {}
for i in range(0, len(tables), 10):  # 10 streams at a time
    group = start_many(spark, options, tables[i : i + 10], trigger={"availableNow": True}, **ids)
    failed |= await_all(group)
    # advance the verdict of each table in the group, as in the first example

• More jobs, each with its own driver (below).

One job, or one job per table¶

Many small or quiet tables fit one job. Give a large or critical table a job of its own, where its batches and its failures touch nothing else. Both can run side by side, as long as no table is in two of them.

On Databricks¶

start_many has not run on Databricks yet; what Running on Databricks says about runtime, access mode, install and network applies to every stream.

• Put the checkpoints in a Volume (/Volumes/<catalog>/<schema>/<volume>/checkpoints/{ci}): the metrics then default under each checkpoint, and on_data_loss="resnapshot" works.
• In a notebook, end the cell that starts the streams with await_all(queries, timeout=...) and call stop_all(queries) in the next cell.
• In Lakeflow Jobs, run start_many in one task, or one stream per table with a For each task over the table list; that is the one-job-per-table column above, with a concurrency limit.

Pitfalls¶

• Never run two jobs over the same list at once: each checkpoint takes one query at a time (Spark refuses a second one in the same session).
• The reader needs SELECT on the change table of each capture instance (Permissions).
• start_many names each query after its app_id: two queries with the same name cannot run in one session.

See also¶

• Streaming into Delta: one stream, which each of these is.
• Monitoring: the facts every stream writes.
• start_many, await_all and stop_all in the API reference.