Processing · Chapter 14

Real-Time and Change Data Capture

Ingestion covered choosing a mechanism and running Auto Loader; this chapter takes the harder case, where the source is an operational database or a stream of events and the business wants freshness measured in minutes rather than hours.

Decide the latency tier before anything else #

Batch, micro-batch, and continuous streaming form a ladder of cost and complexity, where every rung up buys freshness in exchange for operational attention. We ask teams to name the decision that changes when the data is fresher, because the answer usually lands one or two rungs below the initial ask.

Daily is a scheduled job with no long-lived compute, no checkpoint drift, and a full day to recover a failure before anyone notices. It stays right for finance reporting, regulatory submissions, and anything a human reads once a morning.

Hourly is the same architecture on a tighter schedule, where the meaningful change is that a failed run has to be noticed and retried inside the hour, so alerting must work and someone must be accountable during working hours.

Minutes is where micro-batch streaming earns its place. You now own checkpoints, offsets, state, and the question of what happens when a source goes quiet, and compute runs most of the time rather than briefly.

Seconds is a genuine engineering commitment, implying continuous compute, on-call ownership, state stores that need tuning as volumes grow, and consumers able to handle data changing while they read it. We build it where it drives an automated action such as fraud interdiction, and we resist it where the destination is a dashboard somebody opens twice a day.

Managed CDC connectors #

Lakeflow Connect provides managed change data capture connectors for MySQL, PostgreSQL, Microsoft SQL Server, and Oracle. SQL Server can be read either from its change data capture feature or as a full snapshot, and Oracle reads changes through LogMiner. Where one of these covers your source it is almost always the right choice, because the parts most likely to break become Databricks' problem rather than yours.

The architecture has five components, and knowing them is what makes the failure modes legible.

A Connection holds the source credentials as a Unity Catalog securable object, so access is granted and audited like any other object rather than living in a job configuration.

The Ingestion Gateway connects to the source and extracts snapshots, change logs, and the metadata describing them. It runs continuously on classic compute, which makes it the one piece with a cluster you size yourself.

Staging Storage is a Unity Catalog volume holding extracted data before it is applied, purged automatically after thirty days, so it is a buffer rather than an archive.

The Ingestion Pipeline moves data from staging into the destination on serverless compute, scaling with the change volume.

Destination Tables are streaming tables in Unity Catalog, so everything downstream treats them like any other table.

Sources without a managed connector #

Plenty of important sources have no managed connector, and Azure Cosmos DB is the one clients ask about most often, because it sits underneath so many production applications.

Cosmos DB maintains a change feed for each container, an ordered and persistent record of changes to the items in it. The Spark connector, azure-cosmos-spark, exposes that feed as a Structured Streaming source through spark.cosmos.changeFeed options, so the read looks like any other stream. We land it in a bronze streaming table as received, then fold the changes into silver with AUTO CDC.

(spark.readStream
  .format("cosmos.oltp.changeFeed")
  .option("spark.cosmos.accountEndpoint", endpoint)
  .option("spark.cosmos.database", "orders")
  .option("spark.cosmos.container", "order_events")
  .option("spark.cosmos.changeFeed.startFrom", "Beginning")
  .option("spark.cosmos.changeFeed.mode", "AllVersionsAndDeletes")
  .option("spark.cosmos.changeFeed.itemCountPerTriggerHint", "50000")
  .option("spark.cosmos.throughputControl.enabled", "true")
  .option("spark.cosmos.throughputControl.targetThroughputThreshold", "0.25")
  .load())

Two constraints matter more than anything else there. The change feed in its standard latest-version mode gives you the most recent version of each changed item rather than every intermediate version, so an item updated five times between reads arrives once carrying its final state. That is fine for a silver table of current state and wrong if you are reconstructing a full audit history.

Reading the change feed consumes request units on the source container, so this ingestion has a real cost on the Cosmos side as well as on Databricks. The connector's throughput control settings cap that consumption, and we treat them as mandatory, because an unbounded backfill will starve the application depending on that container.

Message brokers #

Kafka, Azure Event Hubs through its Kafka-compatible endpoint, Amazon Kinesis, and Google Pub/Sub are all first-class Structured Streaming sources, differing mainly in connection options.

The mechanics are the same across all of them. The checkpoint holds the offsets recording what has been consumed, so it determines behaviour on restart and belongs with the target table rather than in a scratch location. Broker retention sets how long a stopped pipeline can stay stopped before data is lost, and it should be compared against your worst realistic outage.

The setting that most often changes the bill is Trigger.AvailableNow, which processes everything currently available and then stops. You keep streaming semantics, offsets, and exactly-once bookkeeping while paying for compute only while the batch runs, which is streaming code on batch economics and suits an event stream feeding an hourly report.

Applying changes with AUTO CDC #

Once changes are in bronze, turning that log into a table of current state is done declaratively with AUTO CDC rather than by hand.

CREATE OR REFRESH STREAMING TABLE silver_orders;

CREATE FLOW orders_changes AS AUTO CDC INTO silver_orders
FROM STREAM(bronze_order_changes)
KEYS (order_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY (commit_ts, log_position)
COLUMNS * EXCEPT (operation, commit_ts, log_position)
STORED AS SCD TYPE 1;

The keys identify the row, the sequencing expression establishes the true order of changes, and the delete condition marks removals. Changing the last clause to SCD TYPE 2 produces a full history with validity ranges from the same input.

We insist on this rather than a hand-written MERGE because of out-of-order arrival, which happens more often than teams expect thanks to broker partitioning, retries, and multi-node sources. A merge applying whatever arrived last will silently overwrite a newer value with an older one, and the table looks entirely plausible afterwards, which is what makes the bug expensive.

Datastream and DMS are genuinely good at capture. The difference is what happens after capture, because elsewhere the apply step is code somebody writes and owns, whereas here it is a declarative flow with ordering and deletes handled by the framework.

Operating stateful streams #

A stream that filters rows or renames columns remembers nothing between micro-batches, so its memory profile stays flat for years. That stops the moment you group by a customer, count over a window, deduplicate, or join two streams, because the engine must now hold something about every key it has seen to answer correctly when the next batch arrives. That memory is the state, it lives alongside the checkpoint rather than in the table, and unbounded growth in it is the characteristic failure of the seconds tier. It never announces itself, because the pipeline keeps succeeding while each batch takes slightly longer than the last, until one misses the trigger interval and the lag stops recovering.

A watermark is how you tell the engine it may forget. Setting one with withWatermark on an event time column declares a bound on how late a record may arrive and still be counted, and Databricks is careful about what that promises, in that records arriving inside the threshold are always processed while records arriving outside it may still be processed but are not guaranteed to be. Once the largest event time seen moves far enough past a window, the state for that window can be dropped and its result emitted, which makes state size a function of the watermark rather than of how long the stream has run. What you pay is completeness, because a row later than the bound is dropped or, in a design that cares, routed to a side table. We pick the threshold from the observed lateness of the source rather than from a round number that felt generous, and we watch numRowsDroppedByWatermark, so what we chose to sacrifice stays visible.

from pyspark.sql.functions import window, sum, expr

orders = (spark.readStream.table("silver_orders")
    .withWatermark("order_ts", "30 minutes"))

shipments = (spark.readStream.table("silver_shipments")
    .withWatermark("ship_ts", "2 hours"))

hourly_value = (orders
    .groupBy(window("order_ts", "1 hour"), "region")
    .agg(sum("order_value").alias("region_value")))

# Both sides watermarked, and the join bounded in time, so state can be released.
fulfilled = orders.join(shipments, expr("""
    order_id = shipment_order_id
    AND ship_ts >= order_ts
    AND ship_ts <= order_ts + INTERVAL 2 HOURS
"""))

The state is held per key by a state store committed with every batch, and the provider matters as soon as the keyspace is wide. RocksDB is the production answer, because it keeps state on local disk with an in-memory cache rather than on the JVM heap, which is what stops a large keyspace from turning into garbage collection pauses. It is the default from Databricks Runtime 17.3 onwards and is selected below that through spark.sql.streaming.stateStore.providerClass. Two numbers deserve a dashboard from the first day rather than the first incident. One is state size, reported per stateful operator as numRowsTotal and memoryUsedBytes in the streaming progress metrics, where a line that never flattens means no watermark is doing its job. The other is batch duration against the trigger interval, because a stream is healthy while batches finish comfortably inside their trigger and is in trouble when they creep towards it.

Stream-stream joins are the strictest case, needing a watermark on both sides and a condition bounding the two event times against each other, as in the interval above, with outer joins requiring the watermarks rather than merely benefiting from them. Without both halves the engine can never conclude that a row will not find a partner later, so it keeps every unmatched row for as long as the stream runs, and the join looks fine in testing because a single day of data gives nothing time to go wrong.

For aggregations, deduplication, and joins, Databricks recommends the built-in operators over custom logic and we agree, because that is where watermarks, state cleanup, and recovery are already handled correctly. The escape hatch, transformWithState and its pandas counterpart transformWithStateInPandas, available from Databricks Runtime 16.2, earns its place when the logic really is a state machine rather than an aggregate, meaning named state variables per key, timers that fire on time rather than on the arrival of a row, or a time-to-live on individual values.

Advanced concerns #

Schema drift in the source happens without notice. A column added upstream should reach bronze and surface as an alert, and a column whose type changes should fail loudly rather than being coerced. We keep bronze permissive and silver strict, so drift is captured before it is interpreted.

The initial snapshot plus catch-up is where double counting begins. The snapshot represents a point in time while the change stream starts from an offset, and if the two do not line up you either lose changes in the gap or reapply changes already in the snapshot. Managed connectors handle this, whereas hand-built pipelines need the change stream started before the snapshot is taken and the sequencing column to resolve the overlap.

Idempotent merges matter more than delivery semantics. Teams spend a great deal of energy chasing exactly-once delivery when at-least-once delivery plus an idempotent apply gives the same outcome with far less fragility. If applying the same change twice produces the same table, duplicate delivery stops being an incident.

Backfill alongside a live stream should run as a separate flow into the same target, with the live stream continuing throughout. Stopping the stream to backfill creates a gap somebody has to remember to close, and the sequencing column ensures the backfill cannot overwrite newer records.

Multi-region sources need a decision about where changes are applied, because cross-region egress on a high-volume feed is a meaningful line on a cloud bill, and clock skew between regions makes a wall-clock sequencing column unsafe. We prefer a monotonic source-side sequence such as a log position instead.

Lag is the metric that matters, measured from source commit time to the moment the change is queryable in the target, rather than whether the job succeeded. A pipeline can succeed every run while falling steadily further behind, and only lag reveals it. We publish lag per feed against a threshold agreed with the business, as described in Observability.

The declarative framework hosting these flows is covered in Pipelines, and the cost consequences of continuous compute in Cost and Performance.

Table of Contents