Real-time CDC with MERGE and Streaming Data

1. The Challenge with Streaming Updates

When you read a stream of data, you typically use a sink like .format("delta").outputMode("append"). This works perfectly for new, independent records. However, a CDC feed contains updates and deletes for existing records. Simply appending these changes would result in incorrect data, as you'd have multiple versions of the same row instead of the intended modification.

So, how can we apply MERGE logic, which is inherently a batch operation, to a continuous stream of data?

2. The Solution: The foreachBatch Sink

Spark Structured Streaming provides a powerful sink called foreachBatch. This command allows you to "break out" of the streaming paradigm for each micro-batch. It essentially says: "For each small batch of data that arrives in the stream, treat it as a static DataFrame and run this custom function on it."

This is the perfect tool for our CDC use case. We can use foreachBatch to apply a MERGE operation on every micro-batch of change data that flows from our source.

3. Implementing the Basic Streaming MERGE

Let's watch a detailed video that walks through a complete, practical example of this pattern. The presenter will:

1. Set up a source table and a target table.
2. Generate some new data, including updates to existing records.
3. Demonstrate that a standard append stream fails to handle these updates correctly.
4. Implement the foreachBatch pattern with a MERGE statement to properly apply the changes.

Code Structure

As you saw in the video and documentation, the core structure in PySpark looks like this:

# 1. Define a function to perform the MERGE operation on a micro-batch
def upsert_into_delta(micro_batch_df, batch_id):



    # You can use the DeltaTable API
    target_table = DeltaTable.forName(spark, "my_target_table")
    
    (target_table.alias("t")
      .merge(
          micro_batch_df.alias("s"),
          "s.primary_key = t.primary_key"
      )
      .whenMatchedUpdateAll()
      .whenNotMatchedInsertAll()
      .execute()
    )




# 2. Read the source stream (this could be a CDF stream)
source_stream_df = (spark.readStream
                    .format("delta")
                    .option("readChangeData", "true")
                    .table("my_source_table_with_cdf"))




# 3. Apply the function to each micro-batch
(source_stream_df.writeStream
  .foreachBatch(upsert_into_delta)
  .outputMode("update")
  .start()
)

This pattern elegantly combines the continuous nature of streaming with the powerful DML capabilities of batch processing.

4. Building a Robust CDC Pipeline

The basic implementation is a great start, but real-world data is messy. For a professional-grade pipeline, you need to handle several complexities. The documentation below walks through building a more robust solution step-by-step.

Let's break down the key challenges and solutions from that document:

Challenge 1: Multiple updates to the same key in one micro-batch

A source micro-batch might contain both an INSERT and an UPDATE for the same record. A naive MERGE would error out.

• Solution: Before the MERGE, you must deduplicate the micro-batch to keep only the latest change for each key. A common pattern for this is using groupBy() with the max_by() function on a sequence number or timestamp.

# Inside the foreachBatch function
latest_changes_df = (micro_batch_df
    .groupBy("id")
    .agg(max_by(struct("*"), "sequenceNum").alias("latest"))
    .select("latest.*")
)



# Now, use latest_changes_df in your MERGE

Challenge 2: Out-of-order updates across micro-batches

An older update might arrive in a later micro-batch than a newer update. A simple MERGE would incorrectly overwrite the newer data with the older data.

• Solution: Store the sequenceNum in the target table as well. Then, add a condition to your MERGE statement's WHEN MATCHED clause to only update if the source change is newer than the target's current version.

/* Inside the MERGE statement */
WHEN MATCHED AND s.sequenceNum > t.sequenceNum THEN UPDATE SET ...

Challenge 3: Handling deletes

So far, we've only handled inserts and updates.

• Solution: Add another WHEN MATCHED clause to your MERGE statement to handle delete operations, typically identified by a column in the CDC feed (e.g., operation = 'DELETE').

/* Inside the MERGE statement */
WHEN MATCHED AND s.operation = 'DELETE' THEN DELETE

Challenge 4: Out-of-order updates arriving after deletes

This is the most complex case. If you hard-delete a row, and an out-of-order update for that same row arrives later, the MERGE will see it as a new record (WHEN NOT MATCHED) and incorrectly re-insert it.

• Solution: Tombstoning (Soft Deletes). Instead of immediately deleting the row, you "tombstone" it by setting a flag or timestamp (e.g., DELETED_AT = now()).
• Your users then query a view built on top of the table that filters out the tombstoned records (WHERE DELETED_AT IS NULL).
• A separate, periodic cleanup job can then safely hard-delete tombstoned records that are older than a certain threshold.

This tombstoning pattern is a critical technique for building resilient data pipelines that must handle out-of-order data.

Conclusion

You have now learned how to bridge the gap between streaming data and the powerful MERGE command to build robust, automated CDC pipelines. This is a cornerstone skill for any data engineer working on the Databricks Lakehouse Platform.

Key Takeaways:

• The foreachBatch sink is the key to applying batch-like operations (like MERGE) within a Structured Streaming query.
• A basic streaming MERGE handles simple upserts but is not robust enough for production CDC scenarios.
• A professional-grade CDC pipeline must handle duplicates within a micro-batch, out-of-order data, and deletes.
• Key techniques for robust CDC include:
  • Deduplicating micro-batches using groupBy and max_by.
  • Using sequence numbers to prevent overwriting new data with old data.
  • Using "tombstoning" or soft deletes to correctly handle out-of-order updates that arrive after a delete.

Preview of the Next Lesson:
This lesson concludes the "Incremental Data Ingestion" module. We've focused heavily on getting data into Delta Lake efficiently. In our next module, "Building Streaming Data Pipelines," we will take a step back and explore the fundamentals of Spark Structured Streaming in more detail. In the first lesson, you will learn how to read from various streaming data sources and understand the foundational concepts that power these near real-time pipelines.