Options¶

Everything a stream takes: the source options, given once to stream() or one by one to spark.readStream.format("mssql_cdc"), and the parameters of to_delta(), snapshot(), backfill() and seed().

from mssql_cdc import stream

options = {
    "connectionString": "Server=host,1433;Database=db;Encrypt=yes",
    "user": "cdc_reader",
    "password": password,  # from your secret store
    "captureInstance": "dbo_orders",
    "maxCommitsPerBatch": "500",
}
query = stream(spark, options).to_delta(
    "bronze.orders",
    app_id="orders-v1",
    checkpoint="/Volumes/cat/sch/vol/ckpt/orders",
    facts_table="ops.ingestion_facts",
)

Option names are case-insensitive (startinglsn works). stream() logs one WARNING naming any option it does not know, in your job's log, which is how a misspelt name shows; the reader logs it again in the driver's stderr log, also with spark.readStream, and the facts row of its first batch keeps it (Monitoring). Pass values as strings, as Spark does. A boolean option is true for true, 1, yes or y and false for false, 0, no or n, in any case; anything else raises ValueError naming the option, so a typo such as failOnDataLoss=ture never turns a guard off.

For a type checker to flag a misspelt name before anything runs, annotate the dict as a SourceOptions, which knows every source option below with this spelling and a string value (options: SourceOptions = {...}). It is a plain dict at run time, and every function that takes options takes any mapping too.

Source options¶

"Snapshot" is spark.read.format("mssql_cdc_snapshot"), which snapshot(), to_delta(bootstrap=True), the re-snapshots, backfill() and reconcile() use.

captureInstance¶

The CDC capture instance to read, such as dbo_orders (SQL Server's default name is <schema>_<table>). Any name SQL Server allows, Unicode letters included (dbo_Café), except one with ] or a control character. Matched ignoring case, as SQL Server does; an exact match wins, and two instances whose names differ only in case fail with a ValueError asking for the exact one.

When the table gets a newer capture instance the stream moves to it on its own, and a configured name that was disabled still resolves when it is the table's default name. See Schema changes.

connectionString¶

Connection string for mssql-python or ODBC Driver 18, for example Server=host,1433;Database=db;Encrypt=yes; backend=arrow-odbc adds Driver={ODBC Driver 18 for SQL Server} when it names no driver. Not needed with backend=fake. The driver plans with it and every executor task opens its own connection with it, so the server must be reachable from every worker. The login needs the grants in Permissions.

• Give the login in options of its own rather than as UID and PWD here: user and password, or an Entra ID access token (accessToken, tokenProvider). The driver escapes them, and Spark redacts password and accessToken in a query's plan, which it does not do for this one (under Spark Connect, see password).
• A value written here that contains ; or } goes in braces with every } doubled: PWD={p;a}}ss} for p;a}ss.
• Entra ID authentication by the driver itself, where the server takes Entra ID logins (Azure SQL, SQL Server with Entra ID): Authentication=ActiveDirectoryMsi uses a managed identity and needs no secret at all; Authentication=ActiveDirectoryServicePrincipal takes the application id as user and its secret as password.
• Spark does not redact this option by default: its default patterns (secret, password, token, url...) match neither the key connectionString nor PWD=, and a streaming query's plan shows it whole. Where options can surface (a table defined with OPTIONS (...), a plan in the UI or a log), keep the secret out of it, or set spark.sql.redaction.options.regex to (?i)url|connectionstring.

user¶

The login's name, passed to the driver apart from the connection string: mssql-python takes it as its UID keyword argument, arrow-odbc as arrow_odbc.connect(user=...), and each escapes it, so it needs no braces. With Authentication=ActiveDirectoryServicePrincipal in the connection string it is the application (client) id; with Authentication=ActiveDirectoryMsi, a user-assigned identity's client id. A UID in connectionString as well is a ValueError: give it once.

password¶

The login's password, or a service principal's secret, passed to the driver apart from the connection string as user is: no braces, no doubled }. Read it from your platform's secret store:

options = {
    "connectionString": "Server=host,1433;Database=sales;Encrypt=yes",
    "user": "cdc_reader",
    "password": secret("mssql-password"),  # your secret store's call
    "captureInstance": "dbo_orders",
}

A PWD in connectionString as well is a ValueError. Spark redacts an option named password in a query's plan (spark.redaction.regex, by default (?i)secret|password|token|access[.]?key). No error message of the library holds it: a driver error that does (and SQL Server's "Login failed for user '...'", which names the user) is raised again as a ConnectionError with each value of these options replaced by ***.

Under Spark Connect (Databricks Connect, Databricks standard access mode and serverless) the plan is not the only place: the server also records the text of each request it runs, options included, in the Connect tab of its Spark UI and in its event log, and redacts that text only with the session's spark.sql.redaction.string.regex, unset by default. Anyone who can open that UI or read those logs then sees password, accessToken, user and connectionString as given. There, log in with tokenProvider or a managed identity (Authentication=ActiveDirectoryMsi), which put no secret in the options, or set the server's Spark config to redact them in the request's text:

spark.sql.redaction.string.regex (?i)key: "(password|accesstoken)"\s*value: "(?:[^"\\]|\\.)*"

accessToken¶

A Microsoft Entra ID access token for SQL Server (scope https://database.windows.net/.default), for a server that takes Entra ID logins. mssql-python passes it as the SQL_COPT_SS_ACCESS_TOKEN pre-connect attribute. Every connection, the executors' included, uses this same token, and a token expires: it suits a job that ends before it does. A stream needs tokenProvider.

It replaces the login, so each of these next to it is a ValueError: tokenProvider, user, password, and UID, PWD, Authentication or Trusted_Connection in connectionString. backend=arrow-odbc cannot pass a token (arrow-odbc sets no attribute before it connects), so it is a ValueError there too: use an Authentication= keyword of ODBC Driver 18 instead. Spark redacts it in a query's plan, as it does password, and shows it under Spark Connect as password says.

tokenProvider¶

A function, named package.module:function, that returns an Entra ID access token as a string. The library imports it and calls it with no arguments each time it opens a connection, on the driver and in every executor task, and logs in with what it returns, as accessToken. Only the name crosses to the executors, so the module must be importable on every node, installed with the job's libraries; a stream gets a fresh token each time it connects. How the function gets the token is yours: with azure-identity, which you install (the library does not depend on it), for instance:

# sql_token.py, installed on every node
from azure.identity import DefaultAzureCredential

_credential = DefaultAzureCredential()  # one per Python process, which caches its token


def token() -> str:
    return _credential.get_token("https://database.windows.net/.default").token

options = {
    "connectionString": "Server=myserver.database.windows.net;Database=sales;Encrypt=yes",
    "tokenProvider": "sql_token:token",
    "captureInstance": "dbo_orders",
}

The same options exclude it as accessToken, and backend=arrow-odbc too. A name that does not import is a ValueError; a function that raises fails the connection with a RuntimeError that names it; one that returns anything but a non-empty string (an AccessToken instead of its .token, say) is a TypeError. The messages name the provider, never a token.

backend¶

Anything else fails with ValueError: Unknown backend. Why mssql-python is the default: Connectors and ADR 0003.

connectTimeout¶

Login timeout in seconds, passed to mssql_python.connect, or to arrow_odbc.connect with backend=arrow-odbc. A negative value raises ValueError.

lockTimeoutMs¶

How long, in milliseconds, a read of the source table waits for a lock before it fails with SQL Server's error 1222 ("Lock request time out period exceeded"), sent as SET LOCK_TIMEOUT before each one: the snapshot's reads, a chunked snapshot's planning (to_delta, backfill()) and reconcile()'s counts and ranges. Spark retries a task that fails this way, and planning raises the error to its caller, so a lock held too long fails the job instead of leaving it waiting with no error. With mssql-python, a read that has already returned rows when it times out fails with Unknown DDBC error, which does not name the cause.

Not set, a read waits as long as the lock is held (SQL Server's default). A non-negative integer (0 fails at the first lock met); anything else raises ValueError. The change table reads ignore it. Under isolationLevel=snapshot a read takes no row locks, so it seldom waits (ADR 0029).

sourceTimeZone¶

Time zone of the SQL Server clock. cdc.lsn_time_mapping stores commit times without a zone; this converts them to UTC for _commit_ts, the offsets and the facts. It does not touch captured datetime columns.

• auto: CURRENT_TIMEZONE_ID() on SQL Server 2022+ and Azure SQL, which applies the daylight-saving rules in force at each commit. Older versions lack the function; then the driver reads the server's current UTC offset (SYSDATETIMEOFFSET()) for each batch and applies it to the batch's commits, on the driver and the executors alike, logging a warning when it takes the fallback and when the offset changes, which the facts row of the batch keeps too (Monitoring). That is exact only for zones without daylight saving: a batch that spans a transition, or reads commits from before it, converts some with the wrong offset, and finalized_until can run ahead of the data.
• A Windows time zone name, such as E. South America Standard Time, or UTC: set it on a server older than 2022 in a zone with daylight saving. AT TIME ZONE takes the name since SQL Server 2016 and applies each commit's own offset. In the hour a fall-back repeats, the commits after the clock went back get the offset after the change, so commit times follow commit order.

ADR 0008.

fakePath¶

With backend=fake, the directory that holds the simulator's state (mssql_cdc.fake). Required there, ignored otherwise. Internal, like the fake itself.

columns¶

Spark DDL of the captured columns to read, for example order_id INT, status STRING, amount DECIMAL(18,2). When omitted, the driver infers them at load() from sys.sp_cdc_get_captured_columns, the union of every capture instance of the table, computed columns left out (Output schema).

Pass it to read a subset of the columns, to choose other types (the read casts to them), or when the backend has no column metadata (the fake without captured columns). A listed column that no capture instance of the table captures fails the first planning and the snapshot with a ValueError. A listed computed column reads NULL in every row, snapshot rows too, since CDC stores NULL for it in every change row; the first planning logs a warning naming it, which the batch's facts row keeps (Monitoring). A type change on the source made while the query runs is caught when the batch is planned; one made while it is stopped fails the cast at read time: update columns (Schema changes).

includeCommandId¶

Read __$command_id into _command_id, the order of a statement within its transaction. With false the column is left out of the output. Set it only when the change tables lack __$command_id, which can happen on SQL Server 2016 (ADR 0023). apply_changes works without the column: the order within a transaction then follows __$seqval, which Microsoft's documentation says not to order by (ADR 0009).

startingLsn¶

Where a new checkpoint starts. A checkpoint that already has offsets ignores it.

• earliest: the oldest change CDC still holds (sys.fn_cdc_get_min_lsn of the table's oldest capture instance).
• latest: sys.fn_cdc_get_max_lsn() when the query starts; only later commits are read. On a capture instance that capture has not reached yet (a quiet database just after sys.sp_cdc_enable_table), the LSN just before the instance's first one instead, as a snapshot does.
• An LSN, such as 0x0000002A000001F00003: treated as already processed, so the first batch reads the commits after it. The 0x is optional, and fewer than 20 hex digits are padded on the left. An LSN that CDC cleanup has already passed fails the first batch with DataLossError.

to_delta(bootstrap=True) sets it to the snapshot's LSN, so passing both is a ValueError. After a re-snapshot (generation 1 and later), to_delta replaces it with that snapshot's LSN.

maxCommitsPerBatch¶

At most this many commits per micro-batch, counted in cdc.lsn_time_mapping. That table is database-wide: idle entries and other tables' commits count too. A batch always ends on a commit boundary. Without it every batch reads up to max_lsn, so the first batch after a long stop or a long snapshot reads the whole backlog in one go: set it to keep that batch bounded. A positive integer: omit the option for unlimited (0 raises ValueError).

It needs Spark 4.2+, or a runtime with its admission control backported, such as Databricks Runtime 18.2+ (Databricks). On older Spark it is ignored, with a warning in the driver's log and the facts (Monitoring). With Trigger.AvailableNow the run reads up to the max_lsn it saw when it started, in batches of at most this many commits.

numPartitions¶

At most how many partitions, each with its own connection to SQL Server, a micro-batch is read in. The stream cuts a batch into commit-aligned LSN ranges holding about the same number of change rows, each about 50,000 or more, so a batch of fewer than about 100,000 change rows is read in one partition (ADR 0015). A snapshot cuts a single integer key into uniform ranges between its MIN and MAX, any other key into ranges of about the same number of rows, each bound a seek of its range's keys from the one before, and reads a table without a unique index in one partition. A chunked snapshot's backfill() reads that many chunks at a time, however many a wave takes.

• auto: the cores of the session that called register() (defaultParallelism; stream() calls it), else the CPU count of the node that plans (Spark Connect, or without register()). ADR 0011.
• A positive integer: set it to cap the load on the source. Anything else raises ValueError.

arrowBatchSize¶

Rows per Arrow record batch fetched from the driver, in the stream and the snapshot. Each batch is cast to the output schema and handed to Spark as it arrives. Batches are also capped near 64 MiB, on the default backend as on arrow-odbc, so a table with large (max), text or xml values reads in smaller batches: the default backend starts each read with one row, then fetches as many rows as fit at the widest row of the batch before, at most twice as many. A positive integer; anything else raises ValueError.

failOnDataLoss¶

When the changes the next batch needs are gone (purged by CDC cleanup, or held only by an older capture instance that was disabled before the stream read them), raise DataLossError. It is checked on the driver before a batch is planned and in every task after its read, since cleanup can run in between.

false skips ahead to what CDC still holds and loses those changes; the skip logs a WARNING naming the capture instance and the LSN range skipped, and finalized_until moves past the gap (Finalization). The skip also writes a data_skipped facts row with the gap, through metricsPath: certain when planning finds it, possible (certain false in detail) when a task finds after its read that cleanup ran meanwhile. Prefer to_delta(on_data_loss="resnapshot"), which recovers with a snapshot and records the gap: see Data loss.

schemaChangePolicy¶

What DDL on the source table inside a batch does.

• classify: a captured column whose new type the query's type no longer holds fails the batch before it reads anything (SchemaChangedError; restart to infer the new type). Other DDL goes on, with a warning and a schema_change facts row.
• fail: any DDL fails the batch. The replayed batch holds the same DDL, so restart once with classify to go past it.

Anything else is a ValueError. See Schema changes.

metricsPath¶

A directory where each partition leaves its metrics (round trip, read time, MB, network wait; the batch's last partition also the retention watermark, capture lag and the commit time of the batch's end), and where the reader leaves each schema change, capture instance switch and skip past purged changes (failOnDataLoss). The sink folds them into the facts table and removes them. Without it the metric columns of the facts table and end_lsn stay NULL, and events only reach the driver log.

The files are internal: their names and contents may change in any release, and only the library reads them. The facts table is the public record of what they carry.

It must be a directory every node can write and the driver can read, one of:

• A path: local, or a FUSE mount such as a Databricks Volume (/Volumes/...). Python's file functions write it.
• A URI pyarrow.fs opens (s3://, gs://, abfss://, hdfs://, file://), written through pyarrow.fs.FileSystem.from_uri. Credentials come from pyarrow's defaults on each node (environment variables, an instance profile or managed identity) or from options in the URI's query (s3://bucket/metrics?region=us-east-1), which every node must have. A URI it cannot open, such as dbfs:/, is a ValueError when the query starts. Tested with file:// URIs and a filesystem that cannot rename, not against a real object store yet.

Each file is written whole: under a temporary name moved over its own or, where the filesystem cannot move one, under its own name in one upload. A file that cannot be written is skipped, with a warning in the executor's log: metrics never fail a read.

• Through to_delta, an explicit metricsPath holds each stream's files under <metricsPath>/<app_id>, so streams may share it. Without one, to_delta uses <checkpoint>/_mssql_cdc_metrics (of the live generation) when it has a facts_table or snapshot_on_switch=True and the checkpoint is not a URI. A URI checkpoint (abfss://, dbfs:/) gets no default: set it, to a Volume or to a URI as above.
• By hand, pass the same directory to delta_sink(metrics_path=...) and use it for one stream only: the sink folds every file in it, and nothing else removes them.

What the metrics become: Monitoring and Tables.

snapshotLsn¶

For mssql_cdc_snapshot only: the LSN stamped on the snapshot's rows. By default, sys.fn_cdc_get_max_lsn() recorded before the table is read (or, when capture has not reached a just-enabled instance yet, the LSN just before its start). snapshot() and to_delta set it themselves. Set it only for a snapshot you read yourself, with an LSN recorded before the read: a row in the snapshot may be newer than the LSN, never older (ADR 0016). A chunked snapshot's waves are stamped this way by backfill(), never below the snapshot's own LSN.

snapshotChunks¶

For mssql_cdc_snapshot only: read only these chunks of the table, one partition each, as backfill() and reconcile() do. A JSON list of [chunk, lo, hi]: the chunk's number, its first key (inclusive) and the key it ends before (exclusive). A key is a value for a one-column key and a list for a composite one, null for an open end; a value JSON has no type for is the text CAST reads back (a datetime in ISO 8601, binary as 0x hex), as backfill() plans them. With it, the rows gain a _chunk INT column with the chunk's number, and with metricsPath each chunk leaves its read metrics there, for backfill() (internal files, as that option's are). Key bounds on a table whose key cannot be read in ranges (no unique index, a type no bound can be bound as) are a ValueError. backfill() sets it; set it yourself only to read part of a table.

snapshotKeys¶

The columns a snapshot cuts its ranges on, as a JSON list such as ["created_at"], instead of the capture instance's unique index; anything but a non-empty list of names is a ValueError. They need not be unique. A table without a unique index is otherwise read as one range, and one chunk. Names match the captured columns ignoring case.

• mssql_cdc_snapshot: key bounds (snapshotChunks, and the ranges of numPartitions) apply to these columns. reconcile() sets it to the key it compares, and backfill() to the chunked snapshot's columns.
• to_delta(..., snapshot="chunked"): the snapshot it opens cuts its chunks on these columns, recorded with it in the snapshot plans table, so its plan never changes. They must be captured columns of a type a bound can be bound as (not time, nor a datetime2(7) or datetimeoffset(7) before the last one) and SQL Server can sort (not bit, text, ntext, image or xml), or the open raises ValueError. Each chunk is a range [lo, hi) of their values, NULL first, so a row is in one chunk only, and a value is never split: one held by more rows than chunk_rows makes a bigger chunk. Keep the option while the snapshot is open: backfill() raises when the stream's options would cut on other columns (Tables without a unique index).

On a column no index leads, each range, and each bound the chunked plan seeks, scans the table.

isolationLevel¶

For mssql_cdc_snapshot only: readCommitted (the default) or snapshot, in any case, which reads under SNAPSHOT isolation and needs the database's ALLOW_SNAPSHOT_ISOLATION; SQL Server refuses it otherwise. A SNAPSHOT read does not wait for writers' locks and holds versions in tempdb while it runs. Anything else is a ValueError: a snapshot never reads uncommitted rows (NOLOCK). backfill() reads and plans with the stream's isolationLevel unless its isolation= argument, which takes the same values, says otherwise.

to_delta parameters¶

stream(spark, options).to_delta(
    target,
    app_id,
    checkpoint,
    facts_table=None,
    *,  # the rest by keyword only
    trigger=None,
    query_name=None,
    bootstrap=False,
    on_data_loss="fail",
    resnapshot_interval_days=7.0,
    snapshot_on_switch=False,
    snapshot="full",
)

It starts the query and returns its StreamingQuery (StreamingQueryLike: a Spark Connect one on a Spark Connect session). stream(spark, options) registers the data source on the session first. The parameters after facts_table are keyword-only; on_data_loss and snapshot are typed OnDataLoss and SnapshotMode (Types). A guide to all of it: Streaming.

target¶

The bronze Delta table: a table name such as bronze.orders, or a path (anything with a / or a :). Created on the first batch with rows (Tables).

app_id¶

The sink's identity: the Delta txnAppId of every append to target, with the batch id as txnVersion, so a replayed batch is skipped. Keep it stable for the life of a checkpoint. A new checkpoint needs a new app_id: its batch ids restart at 0 and would be skipped as already written. With a facts_table, a checkpoint that was deleted or rewound under the same app_id fails its first batch with an error saying so, instead of skipping the writes; without one, nothing catches it. Facts rows use <app_id>#facts and snapshot events <app_id>#events; generation n of a re-snapshot writes as <app_id>.g<n>.

checkpoint¶

The Spark checkpoint location; generation n uses <checkpoint>/_generations/<n>. A local path or a Volume lets to_delta place the metrics there and is required by on_data_loss="resnapshot". A URI (abfss://, dbfs:/) works for everything else, with metricsPath set by hand. Not /dbfs/... with on_data_loss="resnapshot": Spark and Python resolve it to two different directories.

to_delta's bootstrap and on_data_loss="resnapshot" pre-flight, seed() and backfill() run in the calling Python process: they connect to SQL Server, and the pre-flight reads the checkpoint, from there. From a Spark Connect client (Databricks Connect included), that machine must reach SQL Server and see the checkpoint path. When the facts table already holds batches of the stream but Python finds nothing of Spark's in the checkpoint, the pre-flight raises ValueError instead of re-snapshotting: either Python and Spark resolve the path to two different directories (on Databricks classic, a path without a scheme such as /mnt/... is DBFS for Spark and the driver's disk for Python; a Spark Connect client's own disk), so use a path both see, such as a Volume; or the checkpoint was deleted, so start again with a new app_id.

facts_table¶

The facts table, a name or a path: one row per micro-batch, snapshot and source change (Tables). Streams may share one. None writes no facts. Required by on_data_loss="resnapshot".

trigger¶

Keyword arguments for DataStreamWriter.trigger, such as {"availableNow": True} or {"processingTime": "1 minute"}. None is Spark's default: the next batch as soon as the previous one ends and the source has moved, which polls SQL Server back to back while the source is idle. Every batch writes a facts row, so a processingTime trigger bounds both for an always-on stream (Triggers).

query_name¶

Passed to DataStreamWriter.queryName. By default the sink's app_id, <app_id>.g<n> in generation n, so the Spark UI, progress logs and listener events match the facts rows.

bootstrap¶

True appends a snapshot of the source table to target before the first batch (once: later runs find it, under any capture instance of the table) and starts the checkpoint at its LSN, so the target holds the whole table, not only what CDC retention still has. With a facts_table it also writes a bootstrap facts row. Not with startingLsn (ValueError). On a table too big to snapshot within the retention, with snapshot="chunked", or after seed: it then finds the seed and reads nothing (Bootstrap).

on_data_loss¶

• "fail": when CDC cleanup purged changes the stream has not read, the query stops with DataLossError.
• "resnapshot": before the query starts, check whether that has happened; if so, snapshot the table into target again, write a resnapshot facts row with the gap, and continue in a new generation. Requires facts_table, failOnDataLoss left true (ValueError otherwise: a purge skipped while the query runs would hide the gap from the next check) and a local or Volume checkpoint; run one job per stream. A purge while the query runs still fails it, and the next run recovers.

Anything else is a ValueError. See Data loss.

resnapshot_interval_days¶

With on_data_loss="resnapshot": at most one automatic re-snapshot, or failed attempt, per this many days. A second loss inside the interval raises DataLossError instead, since the stream does not keep up with the retention. Keep it above the CDC retention (3 days by default); 0 allows a retry right away.

snapshot_on_switch¶

True snapshots the table after the batch that first reads a newer capture instance, so that rows unchanged since the switch carry the columns only the newer instance captures instead of NULL. With snapshot="full" it reads the whole table inside that batch, so the stream waits for the read: not for tables too big to snapshot. With snapshot="chunked" it only opens a chunked snapshot at the newer instance's start LSN, which the stream has read past, and backfill() reads it next to the stream; it supersedes a chunked snapshot still open, and a run with snapshot="full" raises until it completes. With a URI checkpoint it needs metricsPath (ValueError otherwise). See Schema changes.

snapshot¶

How bootstrap=True, on_data_loss="resnapshot" and snapshot_on_switch=True take a snapshot.

• "full": read the whole table before the query starts, as above.
• "chunked": only open one: record its LSN S in a snapshot_open facts row, and the key's extent in the snapshot plans table next to target, then start the stream (the new generation, after a loss) at S at once. backfill(), run in a task of its own, plans the chunks, reads the table in chunks next to the stream and writes the bootstrap or resnapshot row at the end. Requires facts_table (ValueError). A rerun finds the snapshot it opened and opens no second one.

Anything else is a ValueError. The mode holds for the whole run, its bootstrap and its re-snapshot, and may change between runs. With a facts_table, both modes record the snapshot as open before reading, and a run in one mode raises ValueError while a snapshot of the other mode is still open, saying how to finish it; a full one stops counting once CDC cleanup passes its LSN, as it can then never complete (One mode per run). See Bootstrap and ADR 0028.

backfill parameters¶

status = stream(spark, options).backfill(
    "bronze.orders",
    app_id="orders-v1",
    facts_table="ops.ingestion_facts",
)
# {"snapshot": "0x...", "chunks_done": 8, "chunks_total": 40, "done": False,
#  "paused": False, "state": "running", "reason": None}

backfill(target, *, app_id, facts_table, chunk_rows=None, max_waves=None, max_seconds=None, target_wave_seconds=None, min_headroom_hours=None, isolation=None) reads the newest chunked snapshot that to_delta(..., snapshot="chunked") opened for target, in waves of chunks read numPartitions at a time, and returns how far it got. Call it again until done.

• target, app_id, facts_table: as passed to to_delta; the stream's generations (<app_id>.g<n>) are found from app_id. With a full snapshot of the stream still open it raises ValueError, before anything else (One mode per run). Otherwise, without an open chunked snapshot it returns at once, paused with state no_snapshot and a reason.
• chunk_rows: the most rows a chunk holds when planned (at least 1; None is 1,000,000). It counts on the first call only, which plans every chunk and records the plan in the snapshot plans table and a snapshot_plan facts row (its chunk count, no bounds); later calls keep the plan's value and log a warning when given another. One integer key is counted per slice on the server and the slices packed into chunks of at most chunk_rows rows, each but the last short of it by less than one slice (about a sixteenth of chunk_rows on an even key); other keys are cut every chunk_rows keys. The last chunk ends just above the MAX recorded at the open (for other keys at the first key after it, looked for again before each wave while there is none): rows inserted above it come from the stream (How chunks are sized). The chunks are cut on the columns the snapshot opened with, the unique index's or snapshotKeys: a stream whose options would cut on others raises ValueError.
• max_waves, max_seconds: stop after that many waves, or before a wave once that many seconds have passed. None: until the snapshot is done.
• target_wave_seconds: how long a wave should take (at least 0; None is 300). A wave takes whole rounds of numPartitions chunks: one round when the call starts with no wave read yet, then as many rounds as the target holds at the pace of the last wave read (an earlier call's, from its facts rows), and no more than what is left of max_seconds holds. 0 reads one round per wave, as before 0.5.0. Only how many of the plan's chunks a wave takes changes: the chunks, their order and chunk_rows do not.
• min_headroom_hours: before each wave, pause while the stream's newest facts row has less retention headroom (less that row's age), or there is none. None: never pause.
• isolation: "snapshot" sets isolationLevel and plans the chunks under SNAPSHOT isolation too (the counts and seeks of the first call, and the search for the first key after MAX before each wave), so neither waits for writers' locks; "readCommitted" reads and plans READ COMMITTED. Either in any case; anything else is a ValueError before anything is read. None takes the stream's isolationLevel option.

The result, a BackfillStatus: snapshot (its LSN S), chunks_done, chunks_total (the plan's count, None until a call has planned it), done, paused with its reason, and state, which says what to do next:

Each wave is one commit to target and one snapshot_chunk facts row per chunk, its bounds in the snapshot plans table (Tables); a snapshot opened before 0.7.0 keeps them in its facts rows. The next wave is read while that commit and those rows are written, in a background thread: one wave reading and one committing at most, its rows cached until its commit. Where the platform refuses to cache (Databricks serverless), a wave is read by its commit, one after the other (ADR 0032).

snapshot parameters¶

offset = stream(spark, options).snapshot("bronze.orders")
# {"lsn": "0x...", "commit_ts": "2026-09-28T14:03:12.117"}

snapshot(target, *, resnapshot=False, app_id=None, facts_table=None) appends the source table's current rows to target as operation 0 and returns the offset they are stamped with (an Offset), to start a stream from (startingLsn). When target already holds a snapshot of the table, it returns that one's offset and reads nothing. resnapshot=True always takes a new one. With facts_table (and the stream's app_id, else ValueError), a chunked snapshot of that stream still open in target raises ValueError: this one is full (One mode per run). to_delta(bootstrap=True) and on_data_loss="resnapshot" call it for you.

seed parameters¶

offset = stream(spark, options).seed(
    "bronze.orders", copy_df, as_of, app_id="orders-v1", facts_table="ops.ingestion_facts"
)

seed(target, df, as_of, *, app_id=None, facts_table=None, allow_missing_columns=False, reseed=False) appends df, a copy of the table you already have, to target as its snapshot, for a table too big to snapshot within the CDC retention; to_delta(bootstrap=True) then starts from it.

• as_of: when the copy started being read, as a UTC datetime (an aware one is converted), or an LSN recorded before that. Every commit at or before it must be in the copy.
• app_id, facts_table: with a facts table, the stream's app_id; the seed writes the bootstrap facts row to_delta would, and raises ValueError while a snapshot of that stream, full or chunked, is open in target.
• allow_missing_columns: a captured column df lacks reads NULL instead of raising ValueError.
• reseed: append a copy newer than the snapshot already in target, after a DataLossError; then start a new checkpoint and app_id.

A rerun with the same as_of returns the seed already there. See Bootstrap and CdcStream.seed.