Streaming Change Data Feed for Incremental Downstream Processing

1.Why plain streaming breaks on updates and deletes

A plain Delta streaming source only works on append-only tables. Once the source table takes UPDATE, DELETE, MERGE INTO or OVERWRITE, you have four options, and only one of them carries every change type downstream.

skipChangeCommits keeps the stream alive by ignoring every file-changing commit, so updates and deletes never reach the target. The older ignoreChanges re-emits rewritten files, which means duplicates, and still drops deletes. A full refresh is simple but you pay for the whole dataset every time. Materialized views propagate changes for you, but with higher latency, and they exist only in Lakeflow pipelines and Databricks SQL. The change data feed carries inserts, updates and deletes downstream as explicit events. The cost is that you write logic to handle each _change_type.

Sources12

2.Configuring a readChangeFeed stream

Streaming is the recommended way to consume a change feed, and it is the only way Databricks tracks which versions you have already processed. With a batch table_changes call, keeping track of versions is your job. A stream's checkpoint records it for you.

(spark.readStream
  .option("readChangeFeed", "true")
  .table("myTable")
)

That initial snapshot is right for a new pipeline whose target is empty. If the target already holds everything up to some version, pass startingVersion (or a timestamp) so the stream doesn't load the whole source state again as INSERT events.

The usual rate limits, maxFilesPerTrigger and maxBytesPerTrigger, plus excludeRegex, work on change data. After the initial snapshot, limits apply to whole commits. A micro-batch takes all of a commit or defers it to the next batch, so a large MERGE is never split across batches.

Sources2

3.Retention, permanent history, and checkpoint recovery

The change feed is not an archive. It only holds changes made after it was enabled, and its records are transient. When table versions are cleaned up, their change data goes with them. With legacy CDF, VACUUM deletes the change data files, and changes computed from the transaction log follow the checkpoint retention policy. Once a version is gone you can't read its changes any more. If you need a permanent change history, read the feed incrementally and write it to a table of its own.

For that kind of audit or replay job you rarely need an always-on cluster. An availableNow trigger processes everything available and then stops, so it can run on a schedule:

(spark.readStream
  .option("readChangeFeed", "true")
  .table("source_table")
  .writeStream
  .option("checkpointLocation", <checkpoint-path>)
  .trigger(availableNow=True)
  .toTable("target_table")
)

(spark.readStream
  .option("readChangeFeed", "true")
  .option(" ? ", 76)
  .table("source_table")
  .writeStream
  .option("checkpointLocation", "<new-checkpoint-path>")
  .toTable("target_table")
)

Sources2

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Once CDF is enabled, the table keeps a permanent, queryable record of every change ever made.Why is that wrong?

   The feed only covers changes since enablement, and versions expire. For permanent history, write the feed into its own table.

   Covered in Retention, permanent history, and checkpoint recovery

2. 2.skipChangeCommits lets a stream pass upstream updates and deletes through to the target.Why is that wrong?

   skipChangeCommits drops every commit that rewrites files, so only appends get through. Carrying changes downstream needs CDF or separate logic.

   Covered in Why plain streaming breaks on updates and deletes

3. 3.A new readChangeFeed stream only emits commits made after it starts.Why is that wrong?

   Without a starting version, the first batch is the whole current table emitted as INSERT records. Set startingVersion when the target already has that state.

   Covered in Configuring a readChangeFeed stream

Practise it for real

Create a CDF-enabled table, make inserts, an update and a delete, and see each one appear as row-level change events.

1. 1.Run: CREATE TABLE myschema.t(c1 INT, c2 STRING) TBLPROPERTIES(delta.enableChangeDataFeed=true);

   Why: Legacy CDF has to be switched on per table, here at creation time.

   You should see: DESCRIBE HISTORY shows version 0 as CREATE TABLE with delta.enableChangeDataFeed set to true in its properties.

2. 2.Run the two INSERTs, then UPDATE myschema.t SET c2 = upper(c2) WHERE c1 < 3; then DELETE FROM myschema.t WHERE c1 = 3;

   Why: This gives you one commit of each kind: append, update and delete.

   You should see: DESCRIBE HISTORY shows versions 1–4: WRITE, WRITE, UPDATE, DELETE.

3. 3.Run: SELECT * FROM table_changes('myschema.t', 2);

   Why: Reads every change from version 2 (inclusive) to the latest.

   You should see: One insert at version 2, two update_preimage/update_postimage pairs at version 3, and one delete at version 4.

4. 4.Start spark.readStream.option("readChangeFeed", "true").table(...) on the table with trigger(availableNow=True), writing to a new target with its own checkpoint.

   Why: Shows what a stream emits the first time it runs without a starting version.

   You should see: The target receives the table's current rows as insert records.