Skip to content

mssql-cdc-pyspark

A PySpark streaming source for SQL Server Change Data Capture (CDC), a Delta sink with per-batch facts, and a completeness signal, finalized_until, that tells downstream jobs when a period of data is safe to read. Pure Python on Spark's DataSource V2 API: no JVM connector and no platform-specific APIs. Tested on local Spark 4.2 (the source alone also on local Spark 4.1, without admission control), on a local Spark Connect server and on Databricks classic compute (DBR 18.2, single node); other Spark 4.1+ runtimes and multi-node clusters are untested. Databricks serverless is a Spark Connect platform the library is built for (no cache API, availableNow only), tested on the local Connect server with caching refused, and run on serverless with the fake backend; its SQL Server path needs a network path from serverless to your server, which has not been tried (Databricks). The metrics need metricsPath on a local or FUSE path every node sees, or on a URI pyarrow.fs opens (s3://, abfss://...).

Compatibility

The streaming engine is covered by unit tests and the SQL Server behaviour by integration tests against SQL Server 2022. From 1.0.0 on, only a major release breaks the Python API, and the state a stream leaves behind (checkpoints, table schemas) never breaks without a migration path. Dropping a Python, Spark, Delta or SQL Server version that has reached its end of life is a minor release, announced in the changelog at least one release before; 1.0 supports Python 3.10 and later (ADR 0021).

Versions

Each release has its own pages: pick yours in the version selector. stable is the latest final release; dev follows main, which may be ahead of PyPI. A release candidate installs only with pip install --pre mssql-cdc-pyspark or its exact pin, such as mssql-cdc-pyspark==0.8.0. The changelog says what each release has.

The problem

With CDC, "the hour has data" no longer means "the hour is complete": a commit can reach the lake after its hour has passed, and a job that reads 14:00 at 15:05 may read it incomplete. Watermarks inferred from the event times a pipeline has seen are a heuristic: they cannot account for records not seen yet, and they stall on an idle table.

SQL Server already knows. The capture process writes changes in commit order, sys.fn_cdc_get_max_lsn() is the last commit it has processed, and on an idle database it keeps writing entries so that LSN keeps moving. This library carries that frontier from the source to the consumer. The reasoning is in Design notes.

What it does

  • Streams the change table of a capture instance into Spark, with offsets that are commit LSNs: every micro-batch is a prefix of the source's commit history and ends on a commit.
  • Writes the changes to a Delta bronze table idempotently, with one facts row per micro-batch (counts, LSN range, lag, retention headroom, network metrics).
  • Loads the whole table once with a snapshot taken at a recorded LSN, so the target holds more than what CDC retention still has.
  • Fails loudly when CDC cleanup purged changes the stream still needed, or re-snapshots on its own and records the gap.
  • Advances finalized_until per table, only after the data is committed, never backwards.
  • Optionally keeps a current-state (silver) table, one row per key, from the change log.

The pipeline at a glance

flowchart LR
  subgraph SQL Server
    T[(dbo.orders)] -- transaction log --> CAP[CDC capture job]
    CAP --> CT[(cdc.dbo_orders_CT)]
    CAP --> MAP[(cdc.lsn_time_mapping)]
  end
  subgraph Spark
    SRC[mssql_cdc source<br/>offset = commit LSN + commit time]
    BR[Delta bronze<br/>change log + facts]
    FIN[control table<br/>finalized_until]
    SIL[silver<br/>one row per key]
  end
  CT --> SRC
  MAP --> SRC
  SRC --> BR --> FIN
  BR -. apply_changes .-> SIL
  FIN --> C[downstream jobs<br/>gate on finalized_until]

In code, one stream and one verdict:

from mssql_cdc import finalization, stream

options = {
    "connectionString": "Server=host,1433;Database=db;Encrypt=yes",
    "user": "u",
    "password": "p",  # from a secret store
    "captureInstance": "dbo_orders",
}
query = stream(spark, options).to_delta(
    "bronze.orders",
    app_id="orders-v1",
    checkpoint="/checkpoints/orders",
    facts_table="ops.ingestion_facts",
    trigger={"availableNow": True},
    bootstrap=True,
)
query.awaitTermination()

end = finalization.end_offset_from_progress(query.lastProgress)
finalization.advance(spark, "ops.table_finalization", "bronze.orders", end)

Where to go next