API reference¶
The public Python surface (ADR 0021),
generated from the docstrings. Everything not listed here is internal and may change in
any release. The options a stream takes are in Options. CdcStream is listed
for its methods: it is what stream() returns, and only that call creates one.
Leading arguments are positional (to_delta(target, app_id, checkpoint, facts_table),
apply_changes(spark, bronze, target)); options, flags and modes after them are
keyword-only (since 0.3).
Pipeline¶
stream ¶
The CDC stream described by options (the data source options).
CdcStream ¶
to_delta ¶
to_delta(target: str, app_id: str, checkpoint: str, facts_table: str | None = None, *, trigger: dict[str, Any] | None = None, query_name: str | None = None, bootstrap: bool = False, on_data_loss: OnDataLoss = 'fail', resnapshot_interval_days: float = 7.0, snapshot_on_switch: bool = False, snapshot: SnapshotMode = 'full') -> StreamingQueryLike
Start the stream into target through delta_sink; returns the StreamingQuery,
a Spark Connect one on a Spark Connect session.
The parameters after facts_table are keyword-only.
trigger: keyword arguments for DataStreamWriter.trigger, e.g.
{"availableNow": True}. query_name: the query's name; the sink app_id (with
its generation) when None. bootstrap: snapshot the table first (see
snapshot); a checkpoint that already has offsets ignores the starting LSN.
snapshot: how bootstrap and on_data_loss="resnapshot" take one.
"full" reads the table before the stream starts; "chunked" (needs
facts_table) only opens one at an LSN S and starts the stream there at once, and
backfill(), run apart, reads it in chunks next to the stream (ADR 0028). With a
facts_table the mode may change between runs, but a snapshot of the other mode
still open (not completed) raises.
on_data_loss: "fail" (the query stops with DataLossError) or
"resnapshot" (recover before starting, in a new generation; needs facts_table,
failOnDataLoss true and a checkpoint that is a local or FUSE path, not a URI or
/dbfs/).
resnapshot_interval_days must exceed the CDC retention: a second loss within it
raises DataLossError instead of snapshotting again.
snapshot_on_switch: after the batch that first reads a newer capture instance of the
table, snapshot it, so that rows unchanged since then carry the columns only the
newer instance captures, instead of NULL (ADR 0023). It reads the reader's events from
the metrics directory, so a URI checkpoint needs metricsPath. With
snapshot="full" it reads the whole table inside the batch; with "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).
metricsPath, when given, must be a directory every node sees: executors write it
and the driver reads it. A local or FUSE path, or a URI pyarrow.fs opens with the
credentials every node has (s3://, gs://, abfss://, hdfs://); the
reader raises ValueError at the query's start for one it cannot open (dbfs:/).
backfill ¶
backfill(target: str, *, app_id: str, facts_table: str, chunk_rows: int | None = None, max_waves: int | None = None, max_seconds: float | None = None, target_wave_seconds: float | None = None, min_headroom_hours: float | None = None, isolation: Isolation | None = None) -> BackfillStatus
Read the newest open chunked snapshot of target (to_delta(...,
snapshot="chunked") opens one, and with snapshot_on_switch=True one more after a
capture instance switch) in waves of chunks of at most about chunk_rows
rows, numPartitions at a time, next to the running stream; returns how far it
got. Run it apart from the stream (its own task) and call it again until done.
The first call plans every chunk (one integer key from row counts per slice of its
range, the slices packed into chunks of at most chunk_rows rows whatever the
skew; other keys chunk_rows keys at a time) and records the plan
in target's plans table, <target>_snapshot_plans, then in a 'snapshot_plan'
facts row without its bounds (ADR 0028 Amendment 4); later calls read it, so the
chunks never change while the snapshot is open (but for the end of a keyset plan's
last chunk, the first key after MAX: when there was none at planning, it is sought
again before each wave).
chunk_rows: 1,000,000 when None; a later call's other value is ignored, with a
warning. A full snapshot of the stream still being read raises (one mode while a
snapshot is open).
The chunks are cut on the columns the snapshot opened with: the unique index's, or
the stream's snapshotKeys option, unique or not, which a table without a unique
index needs to be read in more than one chunk (ADR 0028 Amendment 6). A value never
splits across chunks: one held by more than chunk_rows rows makes a bigger chunk.
Options that would cut on other columns now raise.
Each wave is stamped with max_lsn before it is read, at or after the snapshot's
LSN S, appended to target in one commit (operation 0, _snapshot S,
_chunk) and then recorded as one 'snapshot_chunk' facts row per chunk, its bounds
in the plans table first; after the last chunk comes the snapshot's 'bootstrap' or
'resnapshot' row, which tells downstream to rebuild from S. A snapshot opened before
0.7.0 keeps its bounds in its facts rows to its end, and writes no plans table. After
a crash between a wave's append and its facts rows, the rerun's append is skipped by
Delta and its rows are rebuilt from the one committed (from its commit's userMetadata
or, once log cleanup dropped that, from its rows in target): nothing is appended
twice. After a crash between a wave's plans rows and its facts rows, the rerun reads
the chunks those rows hold, whatever numPartitions and the pace say: Delta skips
its plans rows, so its facts rows have to match them. A wave is read while the one
before it is appended and recorded, in a background thread, one wave at a time. Where the
platform refuses to cache a wave (Databricks serverless, ADR 0032), its append reads
it, once the one before is recorded, and its commit's userMetadata has no counts.
A wave takes whole rounds of numPartitions chunks, partition p reading chunks p,
p + numPartitions... one after the other: one round at first, then as many as
target_wave_seconds (300 when None; 0 for one round) holds at the pace of the last
wave read (an earlier call's from its facts row), and no more than the rest of
max_seconds. The plan's chunks and their order never change.
app_id: the stream's, as given to to_delta. max_waves and max_seconds
bound one call. min_headroom_hours: pause before a wave while the stream's
retention headroom (of its newest facts row, less that row's age) is below it, or
the stream has written none: the snapshot shares the link with the stream, and a
stream the retention passes loses the snapshot too. isolation: "snapshot"
reads the chunks and plans them (the first call's counts and seeks, and the search
for the first key after MAX before each wave) under SNAPSHOT isolation, which the
database must allow; "readCommitted" under READ COMMITTED. None: the stream's
isolationLevel option, READ COMMITTED without one. Either is matched ignoring case.
Returns a BackfillStatus, {"snapshot", "chunks_done", "chunks_total", "done",
"paused", "state", "reason"}: chunks_total is the plan's, None until a call has
planned it.
state says why the call returned, reason the same in words: "done",
"running" (max_waves or max_seconds stopped it: call again),
"waiting_headroom" (below min_headroom_hours), "waiting_metrics" (the
stream has written no facts row with retention_headroom_hours) or
"no_snapshot" (no chunked snapshot opened by app_id's stream in target:
not yet, or a wrong app_id, target or facts_table).
snapshot ¶
snapshot(target: str, *, resnapshot: bool = False, app_id: str | None = None, facts_table: str | None = None) -> Offset
Append the tracked table's current rows to target as operation 0 and return the
offset they are stamped with (an Offset), for the stream's startingLsn.
The LSN is recorded before the table is read. Once target holds a snapshot taken
under any capture instance of this table, its offset is returned and nothing is read,
so a rerun cannot skip the changes after it: a chunked one's S too, open or complete,
once its first wave is in (ADR 0028). resnapshot=True takes a new one:
after a DataLossError, with a new checkpoint and app_id; downstream, rebuild from the
newest snapshot.
to_delta(on_data_loss="resnapshot") does all of that itself.
With facts_table (and the stream's app_id), a chunked snapshot of that stream
still open in target raises: it is a full snapshot (ADR 0028).
seed ¶
seed(target: str, df: DataFrameLike, as_of: str | datetime, *, app_id: str | None = None, facts_table: str | None = None, allow_missing_columns: bool = False, reseed: bool = False) -> Offset
Append df, a copy of the tracked table you already have, to target as its
snapshot, and return the offset it is stamped with, an Offset (ADR 0025). 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 ("0x...") recorded before that. Every commit at or before it
must be in the copy; later ones may be, the stream replays them. A datetime maps to
the last commit at or before it, to the second, on the server's clock
(sourceTimeZone). An LSN CDC cleanup has passed raises DataLossError.
df's columns match the captured ones by name, ignoring case; others are dropped.
A captured column df lacks raises ValueError, or reads NULL with
allow_missing_columns=True. With facts_table (and the stream's app_id)
a 'bootstrap' event row records the seed, the one to_delta would write.
A rerun with the same as_of returns the seed already written, also under a newer
snapshot; once cleanup has passed a time as_of (and its cdc.lsn_time_mapping
row), the newest snapshot of the table at or before it. Any other snapshot of the
table in target raises, unless reseed=True and as_of is newer: after a
DataLossError, then start a new checkpoint and app_id from it. With
facts_table, a snapshot of that stream still open in target, in either mode,
raises too.
register ¶
Register format("mssql_cdc") (the stream) and format("mssql_cdc_snapshot")
(the tracked table's current rows) on a SparkSession.
numPartitions defaults to the cores of this session's compute. The data source
plans in a Python worker that has no session, so the count is taken here and rides
along as an attribute of a registered subclass (Spark pickles it by value).
Many tables¶
start_many ¶
start_many(spark: SparkSessionLike, options: SourceOptions | Mapping[str, Any], capture_instances: Iterable[str] | Mapping[str, SourceOptions | Mapping[str, Any]], *, target: str, app_id: str, checkpoint: str, facts_table: str | None = None, **to_delta_kwargs: Any) -> dict[str, StreamingQueryLike]
Start one to_delta stream per capture instance; returns the queries by capture
instance, in the order given.
target, app_id and checkpoint are templates in which {ci} is the
capture instance, e.g. "bronze.{ci}", "{ci}-v1" and
"/Volumes/main/ops/ckpt/{ci}". Each must contain it: every stream needs its own
target, app_id and checkpoint. capture_instances is a list of names, or a mapping
of name to options for that stream alone, which override options
(numPartitions, columns, startingLsn...). facts_table and the other
keyword arguments (trigger, bootstrap, on_data_loss...) go to every
to_delta; each query is named as to_delta names it, after its sink id
(<app_id>.g<n> after a re-snapshot), as its facts rows are.
The facts table is created, or migrated, once before the first start: writers that commit to a new Delta table at the same time conflict. The streams start in order, each after its own bootstrap or re-snapshot, and once the query before it has started its source (or stopped). When a start raises, the queries already started are stopped and the error is raised.
await_all ¶
await_all(queries: Mapping[str, StreamingQueryLike], *, timeout: float | None = None) -> dict[str, Exception]
Wait until every query has stopped, or timeout seconds in all (keyword-only), and
return the error of each query that stopped with one, by capture instance.
It raises nothing for them: raise when the result is not empty, or the run succeeds
with a table behind. is_data_loss and is_schema_changed tell which errors need
a decision rather than a retry. A query still running at the timeout is not in it (see
isActive). With trigger={"availableNow": True} every query stops once it has
caught up; for queries that keep running, spark.streams.awaitAnyTermination()
returns as soon as one stops.
stop_all ¶
Stop every query still running. Each resumes from its checkpoint when started again,
e.g. by start_many with the same templates.
Silver¶
apply_changes ¶
apply_changes(spark: SparkSessionLike, bronze: str, target: str, *, capture_instance: str | None = None, keys: Sequence[str] | None = None, control_table: str, facts_table: str | None = None, options: SourceOptions | Mapping[str, Any] | None = None, granularity: Granularity = 'hour') -> ApplyResult
Bring target up to the capture instance's changes in bronze, a table fed by
that capture instance alone (and the newer ones of its table it switched to). Until the
stream creates bronze, it does nothing but log a warning that names it.
capture_instance: the stream's; without it, captureInstance of options.
keys: the source's key columns; without them, read from the capture instance's
unique index through options (the stream's), which also name the table's other
capture instances: needed once bronze holds a newer one's rows. control_table keeps
the position and gets the verdict; bronze's own verdict must be under the same name
or path.
facts_table: the stream's, needed to see the re-snapshot of an emptied table and
any chunked snapshot (a call fails on chunk rows without it, or when its rows name
bronze otherwise). Without it, the verdict is not advanced.
The parameters after target are keyword-only.
Returns an ApplyResult, {"rebuilt", "applied_lsn", "finalized_until",
"bronze_found"};
bronze_found is False when bronze does not exist (nothing was applied).
Validation¶
reconcile ¶
Is a silver table equal to its SQL Server table? Count by key range, then compare rows.
from mssql_cdc import reconcile
result = reconcile(spark, options, "silver.orders", bronze="bronze.orders",
control_table="ops.table_finalization",
report_table="ops.reconcile_report")
options are the stream's; silver is the table apply_changes builds from bronze.
Nothing is written to SQL Server, and nothing is read with NOLOCK: Tier 1 counts under READ
COMMITTED, Tier 2 reads through the snapshot reader under the isolationLevel option
(READ COMMITTED by default).
- Tier 1, every row: one scan on SQL Server counts the rows and sums the keys (
COUNT_BIG,SUMasdecimal(38,0)) per bucket of the key's range, and Spark the same over silver; adjacent buckets are merged until they holdbucket_rowsrows. Only for a one-column integer or date key, which SQL Server and Spark order alike. Any other key (composite, or a string a collation orders) gets one count of the whole table. - Tier 2, row by row: every MISMATCH bucket and a
sampleof the MATCH ones (for any other key, a sample of ranges of aboutbucket_rowsrows cut byNTILE) are read through the snapshot reader (snapshotChunks, onkeys:snapshotKeys) and joined with silver on the key, comparingsha2(to_json(struct(<captured columns>)), 256)computed by the same Spark function on both sides (not a computed column, which CDC stores as NULL in every change row). MISSING_TARGET: the key only in the source (an insert not applied); MISSING_SOURCE: only in silver (a delete not applied, a stale key); RECORD_DIFF: other values (an update not applied). For any other key silver is joined from the source's rows, which finds only the first and the last: no range Spark cuts of silver is SQL Server's. The source's rows are read once, into a local checkpoint: where the platform refuses one (Databricks serverless, ADR 0032), Tier 2 is skipped with a warning, every buckethashedfalse. - IN_FLIGHT: M is
max_lsnread just before the source is read, E silver'sapplied_lsnincontrol_table, read before the silver version compared, which holds every change up to it (not its newest_start_lsn: a chunk row's stamp can be ahead of the changes applied). A bucket or a key that differs while a change to it (not a snapshot row) after the older of the two is in bronze, or only in the change table yet, is IN_FLIGHT: check it again later. Once the source is read, bronze is pinned and its change table read after bronze's position (its newest change, or the S of a chunked snapshot, which the stream reads from; not a whole snapshot's, whichsnapshot_on_switchtakes above the stream's position) up tomax_lsnread then, through the capture instances the stream reads: the changes the stream has not read yet. Only a commit capture has not harvested yet (its lag, seconds) cannot be seen: a MISMATCH that causes clears on the next run. While a chunked snapshot is open, the keys of the chunks silver lacks are MISSING_TARGET. Equal counts are a MATCH even then. - Chunks, with
facts_table: bronze's newest chunked snapshot (ADR 0028) is checked against its 'snapshot_chunk' facts rows, their bounds in bronze's plans table (<bronze>_snapshot_plans) unless the rows have them (a snapshot opened before 0.7.0), without reading SQL Server. CHUNK_TILING: the chunks leave a gap or overlap (each starts where the one before ended, the first open below and, once complete, the last the plan's final one); CHUNK_ROWS: a chunk's facts row counts other rows than bronze holds of it; CHUNK_STAMP: a chunk stamped below the snapshot's LSN S. The facts are read before bronze, which commits a wave's rows before its facts rows, so a wave still being written is no failure.
Returns a ReconcileResult, {"run_id", "silver_version", "silver_lsn", "source_lsn",
"buckets", "match", "in_flight", "mismatch", "hashed", "failures": {failure_type: rows},
"report"}, report the run's report as a DataFrame, which report_table also gets
(its columns: "Reconcile report" in the tables reference).
reconcile ¶
reconcile(spark: SparkSessionLike, options: SourceOptions | Mapping[str, Any], silver: str, *, keys: Sequence[str] | None = None, bronze: str, control_table: str, facts_table: str | None = None, bucket_rows: int = 1000000, sample: float = 0.01, report_table: str | None = None, sample_seed: int | None = None, seed: int | None = None) -> ReconcileResult
Compare silver with the table the capture instance in options tracks (how: the
Validation guide); returns a ReconcileResult. The parameters after silver are
keyword-only.
keys: the source's key columns; without them, the capture instance's unique index.
bronze: the table silver is applied from; its newer changes, and the change table's
the stream has not read yet, make a difference IN_FLIGHT. control_table:
apply_changes's, for silver's applied_lsn. facts_table: the stream's, to
check the chunks of bronze's newest chunked snapshot. bucket_rows: rows per bucket.
sample: the fraction of the MATCH buckets also compared row by row (0: none, 1: all).
sample_seed: the random seed of that sample. seed: its name before 1.0, deprecated
(it warns, and leaves in 2.0); both at once raise TypeError.
Finalization¶
finalization ¶
Completeness signal ("partition finalization") for CDC-fed tables.
Two layers, in the spirit of Pinterest's partition finalization:
- Facts, per micro-batch: what was written (row counts, LSN and commit-time
ranges). See
mssql_cdc.sink. - Verdict, per table:
finalized_until. Every period strictly before it is complete in the target table and will not receive more source commits. Over a loss gap a re-snapshot recorded (lost_from_ts..lost_to_tsof the facts' 'resnapshot' rows), a change-log table's verdict means only that nothing more will arrive, not that the gap's changes are in it (ADR 0018).
Why the verdict is sound for SQL Server CDC: a batch's end LSN never exceeds
sys.fn_cdc_get_max_lsn(), the last LSN the capture process has processed, and
the capture process writes change rows in commit order. So once a batch ending at
LSN L is committed to the target, every source transaction committed at or
before the commit time of L is in the target. Transactions sharing that exact
commit time may still be pending, which is why the period containing it is not
finalized; truncating to the period start handles that.
Ordering rather than atomicity keeps it safe: write data first, advance the verdict after, and never let it move backwards. If the job dies in between, the verdict lags (consumers wait a little longer); it can never run ahead of the data.
FinalizationListener ¶
Advances finalized_until after the batches of one query run. Create it with
track, which registers it and returns it.
Spark posts a progress event after the batch's data and checkpoint commits, so the verdict
still follows the data (ADR 0005). The listener bus calls back on a thread every listener
of the session shares, so the callbacks only hand the newest progress to one worker
thread, which calls advance. A progress that arrives while an advance runs replaces
the one still waiting: the newest end offset implies the older ones. The worker skips
an end offset that does not move the verdict, so the control table gets a commit per
period, not per batch. advance itself retries a MERGE that loses to another
tracker's on the same control table; a verdict that still fails is logged, and the next
progress tries again. last_error is the error of the last attempt (None once one
succeeds) and failures how many failed in a row: the first of a streak is logged as
an ERROR, then a WARNING at most every 10 minutes until one succeeds. When the run
terminates, the worker applies what is left, stops and removes the listener. On Spark
Connect it leaves it registered, where removing it can hang the client (PySpark 4.2.0),
and it ignores every other run's events.
join ¶
Wait until the query has terminated and its last verdict is written (or failed,
and logged: see last_error). False when timeout seconds pass first
(keyword-only).
advance ¶
advance(spark: SparkSessionLike, control_table: str, table_name: str, end_offset: Mapping[str, Any] | None, *, granularity: Granularity = 'hour') -> datetime | None
Monotonically advance finalized_until for table_name; granularity is
keyword-only.
Call only after the batch that produced end_offset is committed. A MERGE that loses
to a concurrent commit on the control table is retried for up to a minute.
track ¶
track(spark: SparkSessionLike, query: StreamingQueryLike, control_table: str, table_name: str, *, granularity: Granularity = 'hour') -> FinalizationListener
Advance finalized_until of table_name after every batch of query, a started
StreamingQuery that writes it, until the query terminates; returns the listener.
For a stream that keeps running (a processingTime trigger, or the default). Call it
right after starting the query; join() waits for the last verdict once the query
stops. With availableNow, query.awaitTermination() then join(). The control
table is created here, so a wrong name fails now rather than in the worker's log.
granularity is keyword-only.
finalized_until ¶
table_name's verdict in control_table, a naive UTC datetime: every period
that ends at or before it is complete. None before the first verdict.
is_final ¶
is_final(spark: SparkSessionLike, control_table: str, table_name: str, period_end: datetime) -> bool
True when the period ending at period_end (exclusive) is complete: no source
commit in it can still arrive. Over a loss gap a re-snapshot recorded (the facts'
'resnapshot' rows), a change-log table's period is final without the gap's changes.
An aware period_end is taken in UTC, as the verdict is.
freshness ¶
How old table_name's verdict is now, by this machine's clock: its
finalized_until and when it last moved, as ages in seconds, and the chunked snapshot
that holds it, if one does. Reads the control table, without writing to it; the times
and ages are None before the first verdict. docs/guides/monitoring.md has the alert
query that tells a held verdict from a stopped stream or capture.
candidate ¶
candidate(end_offset: Mapping[str, Any] | None, *, granularity: Granularity = 'hour') -> datetime | None
finalized_until implied by a committed batch's end offset.
end_offset_from_progress ¶
end_offset_from_progress(progress: Mapping[str, Any] | None, *, source_index: int = 0) -> Offset | None
Extract the end offset from a StreamingQueryProgress (object or dict); source_index
is keyword-only.
FinalizationListener is listed for its join, last_error and failures. Create it with
track, which also registers it and starts its worker: its constructor is not public.
Sink¶
delta_sink ¶
delta_sink(target: str, app_id: str, facts_table: str | None = None, *, metrics_path: str | None = None) -> Callable[[DataFrameLike, int], None]
Return a foreachBatch function; metrics_path is keyword-only.
app_id must be stable for the lifetime of a checkpoint. If the checkpoint is
deleted, use a new app_id; batch ids restart at 0 and would otherwise be
ignored as duplicates. With a facts table the run's first batch is checked against it:
a batch id below the largest one app_id already wrote there (the checkpoint was
deleted or rewound) raises ValueError instead. Without one nothing checks it.
metrics_path feeds the facts table with each partition's read and network metrics:
the directory of the source option metricsPath, used by no other stream. Its files are removed after each batch,
with or without a facts table; without metrics_path nothing removes them. It also
carries the reader's schema change, capture instance switch and data skipped events to
the facts, and its other warnings to the batch row's detail.
A batch that read rows but found no metrics file there logs a warning, once per run.
Types¶
Exported from mssql_cdc. The mode parameters take a Literal, so a type checker flags a
misspelt mode; a wrong value still raises ValueError naming the allowed ones. The results
are TypedDicts, plain dicts at run time; a minor release may add keys to them, and only a
major release removes or renames one, listed under "Breaking" in the changelog.
types ¶
The types of the public API: the values a mode parameter takes, and what the calls return.
The results are TypedDicts: plain dicts at run time, with keys a type checker knows. A
minor release may add keys to them; only a major release removes or renames one, listed under
"Breaking" in the changelog (ADR 0021).
SnapshotMode
module-attribute
¶
How to_delta(snapshot=...) takes a snapshot: before the stream, or in chunks next to it.
OnDataLoss
module-attribute
¶
What to_delta(on_data_loss=...) does when CDC cleanup purged changes not read yet.
Isolation
module-attribute
¶
The isolation level backfill(isolation=...) reads and plans under (any case at run time).
Granularity
module-attribute
¶
The period finalized_until is truncated to.
BackfillState
module-attribute
¶
Why backfill() returned: see BackfillStatus.
SparkSessionLike
module-attribute
¶
What a spark parameter takes, and get_spark() returns: a classic SparkSession or a
Spark Connect one (Databricks Connect, serverless compute), which is not its subclass. For
type checkers: at run time it is a string.
StreamingQueryLike
module-attribute
¶
What a query parameter takes, and to_delta() returns: a classic StreamingQuery or a
Spark Connect one, which is not its subclass. For type checkers: at run time it is a
string.
DataFrameLike
module-attribute
¶
What a DataFrame parameter takes, and ReconcileResult.report is: a classic DataFrame or
a Spark Connect one. PySpark 4 makes the Connect class a subclass of pyspark.sql.DataFrame,
so code annotated with that still type-checks. For type checkers: at run time it is a
string.
Offset ¶
A position in the change log, as in a stream's offsets: what snapshot() and
seed() return, for the stream's startingLsn.
BackfillStatus ¶
What backfill() returns: how far the chunked snapshot got.
snapshot
instance-attribute
¶
The snapshot's LSN S; None with state no_snapshot.
chunks_total
instance-attribute
¶
The plan's chunks; None until a call has planned them.
state
instance-attribute
¶
done; running (max_waves or max_seconds ran out: call again);
waiting_headroom and waiting_metrics (paused by min_headroom_hours);
no_snapshot (no chunked snapshot of target opened by app_id's stream).
ApplyResult ¶
What apply_changes() returns.
ReconcileResult ¶
What reconcile() returns: a summary of the run, and its report.
silver_lsn
instance-attribute
¶
Silver's applied_lsn read before that version.
source_lsn
instance-attribute
¶
sys.fn_cdc_get_max_lsn() read just before the source was counted.
in_flight
instance-attribute
¶
Buckets that differ while a change to them is in flight: check again later.
failures
instance-attribute
¶
Report rows by failure_type: MISSING_TARGET, MISSING_SOURCE,
RECORD_DIFF, IN_FLIGHT, CHUNK_TILING, CHUNK_ROWS, CHUNK_STAMP.
report
instance-attribute
¶
The run's report rows, in the columns of the reconcile report table, as
report_table gets them.
Freshness ¶
What finalization.freshness() returns: how old a table's verdict is, for alerts.
finalized_until
instance-attribute
¶
The verdict, naive UTC; None before the first.
updated_at
instance-attribute
¶
When the verdict last moved, naive UTC; None before the first.
age_seconds
instance-attribute
¶
Seconds from finalized_until, the last final instant, to now. The first period not
final yet ends one period after it, so its consumers have waited this less the period,
when that is positive. A current stream keeps it under the period (an hour by default)
plus its trigger interval and the idle entries' 5 minutes.
seconds_since_update
instance-attribute
¶
Seconds since the verdict last moved. Small with a large age_seconds: the verdict
moves, the stream is behind. Large too: it stopped moving.
open_snapshot_lsn
instance-attribute
¶
Tables built by apply_changes: the LSN of the chunked snapshot still open, which
holds the verdict until the snapshot completes; None otherwise.
SourceOptions ¶
The source options (docs/reference/options.md), every one optional to a type checker
and a string, as Spark passes them. A plain dict at run time: the functions that take
options take any mapping of them too, and a name in another case (startinglsn)
still works there, though only this spelling type-checks. fakePath is internal, as
the fake backend it serves: it may change in any release.
Protocols¶
The seam a driver plugs into
(ADR 0030), exported from
mssql_cdc. Backend is a typing.Protocol, runtime_checkable: a class with its methods
is one without inheriting, and isinstance checks that it has them (by name; a type checker
checks the signatures). A subclass inherits the methods that have a body. The data source
builds its client from its options, so the backend option names a
built-in backend only. A method added to the protocol is a break for a class that implements
it without inheriting: only a major release adds one, listed under "Breaking" in the
changelog.
Backend ¶
What the SQL Server client needs from a database driver: run a query, return Apache Arrow.
A typing.Protocol (ADR 0030): a class with batches, scalar and close is
one, inheriting or not; a subclass inherits scalar and close. Every parameter is
text, which the T-SQL converts on the server (ADR 0003).
Lsn
module-attribute
¶
An LSN in its canonical form, 0x and 20 uppercase hex digits (ADR 0030), as
normalize, from_int and the client's methods return it. A str at run time, so it
fits anywhere a str does; parameters that take an LSN accept any str in that form.
Payloads¶
The JSON in the facts table's detail and in the userMetadata of a chunked snapshot's
wave commits (Tables), as TypedDicts for what json.loads returns.
The row payloads are exported from mssql_cdc; the nested types are in mssql_cdc.payloads.
Their keys are state: a release only adds one.
payloads ¶
The JSON payloads a later release reads back (ADR 0021): the detail of the facts rows,
and the userMetadata of a chunked snapshot's wave commits in bronze.
TypedDicts, plain dicts at run time, for what json.loads of those columns returns::
d: SnapshotChunkDetail = json.loads(row["detail"])
A release only adds keys to them, never renames or removes one, nor gives one another meaning.
A key added after its payload first shipped is optional here, and a reader takes it with a
default (.get), so rows an older release wrote stay readable. The one exception: since
0.7.0 a chunked snapshot's facts rows leave its key values (keys and plan of
'snapshot_open', chunks of 'snapshot_plan', lo and hi of 'snapshot_chunk') to
the plans table next to bronze, whose detail holds the payload whole (ADR 0028
Amendment 4, ADR 0021): optional here, as a facts row may lack them.
Key values (lo, hi, max) are JSON: a scalar for one key column, a list for a
composite key, None for an open end; a value JSON has no type for (a date, a decimal) is
its text, binary its 0x hex.
SnapshotExtent
module-attribute
¶
The plan of a chunked snapshot's 'snapshot_open' row: what its chunks tile.
SnapshotKind
module-attribute
¶
Why a snapshot is taken: bootstrap the first one of the target (and a chunked one after
a capture instance switch, switch_batch_id), resnapshot one after CDC data loss.
SnapshotOpenDetail ¶
detail of a 'snapshot_open' facts row, written before a snapshot reads the
table; keys and plan only on a chunked one, and since 0.7.0 only in its plans
table's row; switch_batch_id only on one opened after a capture instance switch.
keys
instance-attribute
¶
The key columns its chunks are cut on: the unique index's, or snapshotKeys.
switch_batch_id
instance-attribute
¶
Since 0.8.0, on a chunked one to_delta(snapshot_on_switch=True) opens: the
batch_id of the batch that first read the newer capture instance (its
'capture_instance_switched' row); S is that instance's start LSN, which the stream
has read past. Its kind is bootstrap: no change was lost.
mode
instance-attribute
¶
How it is taken; a reader takes a missing or unknown one as chunked.
generation
instance-attribute
¶
The stream generation it starts (ADR 0018): 0 for a bootstrap; for one opened after a capture instance switch, the generation that read the switch.
lost_from_ts
instance-attribute
¶
A re-snapshot's: the commit time (UTC ISO-8601) of the last offset processed.
lost_to_ts
instance-attribute
¶
A re-snapshot's: the commit time of the min_lsn cleanup had got to.
SnapshotPlanDetail ¶
detail of a 'snapshot_plan' facts row: every chunk of a chunked snapshot,
planned by its first backfill() and fixed while it is open; chunks since 0.7.0
only in its plans table's row.
chunks
instance-attribute
¶
[lo, hi] of every chunk, in order: from lo (inclusive) to hi (exclusive).
SnapshotChunkDetail ¶
detail of a 'snapshot_chunk' facts row: a chunk of a chunked snapshot, read;
lo and hi, the key range it was read in, since 0.7.0 only in its plans table's
row.
SnapshotCompletionDetail ¶
detail of a chunked snapshot's 'bootstrap' or 'resnapshot' row, written
after its last chunk (a full snapshot's has none).
DataSkippedDetail ¶
detail of a 'data_skipped' facts row (ADR 0018): from the driver, certain
true; or, from a task that found cleanup had run while it read a range, certain false
and a reason.
Attributes:
| Name | Type | Description |
|---|---|---|
from |
str
|
The first LSN not read; from a task, its range's first LSN. |
to |
str
|
The |
certain |
bool
|
True when the changes are lost (the driver's check), false when they may be (a task's). |
reason
instance-attribute
¶
Why the task flagged its range, from a task only; text for people, which may change.
CaptureInstanceSwitchedDetail ¶
detail of a 'capture_instance_switched' facts row (ADR 0023), JSON from 1.0.0rc1:
a row written before holds only text's line, as plain text (events.switch_of reads
both).
start_lsn
instance-attribute
¶
The newer instance's start LSN S, where the stream switched: the row's min_lsn.
not_captured
instance-attribute
¶
The columns the query reads that the newer instance does not capture, NULL from S on; empty when none.
text
instance-attribute
¶
old -> new, then the columns no longer captured: the line a row written before
held. Text for people, which may change.
BatchDetail ¶
detail of a micro-batch row (event NULL), set only when it has warnings.
warnings
instance-attribute
¶
The warnings the reader logged on the driver; text for people, which may change.
WaveMetadata ¶
WaveChunk ¶
A chunk in a wave's userMetadata.
rows
instance-attribute
¶
None where the platform refuses to cache the wave (Databricks serverless): its commit is written as it is read, before the rows are counted; its 'snapshot_chunk' facts row has them (ADR 0032).
high_lsn
instance-attribute
¶
max_lsn after the read: informational; None when it could not be read, and where
rows is None.
IntExtent ¶
A chunked snapshot's extent for one integer key, read just after S.
KeysetExtent ¶
A chunked snapshot's extent for any other key, read just after S.
Errors¶
DataLossError ¶
Raised when requested change data was already purged by CDC cleanup.
SchemaChangedError ¶
Raised when the source's schema changed in a way the running query cannot absorb: restart it to re-infer the schema (ADR 0023).
is_data_loss ¶
Whether exc is a DataLossError, or the exception Spark stopped a query with
because of one: what awaitTermination() raises (StreamingQueryException),
query.exception() or await_all returns. Recover with a re-snapshot
(to_delta(on_data_loss="resnapshot")).
is_schema_changed ¶
Whether exc is a SchemaChangedError, or the exception Spark stopped a query
with because of one (as is_data_loss). Restart the query, or change the target first
when the message asks to.
Local Spark¶
get_spark ¶
get_spark(app_name: str = 'mssql-cdc', master: str = 'local[*]', *, delta: bool = True) -> SparkSessionLike
Return the active SparkSession (Databricks, EMR, Fabric...) or create a
local one, with Delta configured when delta-spark is installed (delta,
keyword-only). A local one also gets the session time zone UTC and 4 shuffle partitions.
With SPARK_REMOTE set, a Spark Connect session on that server, which has its own
configuration: master and delta are not used. On Spark Connect (that, or the
platform's own session) it is a Spark Connect SparkSession.