Re: Updating a Delta Table in Delta Live Tables (D...

mark_ott · ‎11-16-2025

To achieve robust, persistent CDC (Change Data Capture)–style updates in Databricks DLT with your scenario—while keeping data_predictions as a Delta Table (not a Materialized View)—you need to carefully avoid streaming joins and side effects across streaming tables in a single DLT pipeline, because of DLT’s constraints on table references and state handling.

Below is a best-practice architecture and pipeline pattern for your use case:

Key Solution Principles

Ingest Predictor and Confirmer streams each to their own staging Delta Tables (no materialized views or joins here).
Make data_predictions a Delta Table with historical CDC semantics (append-only with late-arriving data).
Perform updates via MERGE INTO (upserts) using a batch job or a separate DLT pipeline instead of streaming/SQL views, as true updates on existing Delta Tables from multiple sources in a single DLT pipeline are not well supported.
Retain historical data by enabling Delta Lake time travel and mergeSchema options, so even after pipeline restarts, your table’s history is intact and queryable.

Step-by-Step Pipeline Design

1. Stage Each Stream in Its Own Raw Table

python

@Dlt.table(name="raw_predictor")
def raw_predictor():
    df = (
        spark.readStream
        .format("kafka")
        .options(**KAFKA_OPTIONS_PREDICTOR)
        .load()
        .selectExpr("CAST(value AS STRING) as json_data")
        .select(from_json(col("json_data"), schema).alias("data"))
        # flatten, select, etc...
        .withColumn("etl_ingest_ts", current_timestamp())
    )
    return df

@Dlt.table(name="raw_confirmer")
def raw_confirmer():
    df = (
        spark.readStream
        .format("kafka")
        .options(**KAFKA_OPTIONS_CONFIRMER)
        .load()
        .selectExpr("CAST(value AS STRING) as json_data")
        .select(from_json(col("json_data"), schema_update).alias("data"))
        # flatten, select, etc...
        .withColumn("confirmer_etl_ts", current_timestamp())
    )
    return df

These are persisted, append-only Delta tables—not materialized views.

2. Build/Persist the Gold Table (`data_predictions`) via a Batch Upsert

Create a separate batch pipeline or notebook that periodically runs a MERGE INTO operation to keep data_predictions current with the latest from both staging tables, using a uuid key.

Example batch upsert job:

python

# in a notebook or DLT batch pipeline cell

# Load tables
df_predictor = spark.read.table("raw_predictor")
df_confirmer = spark.read.table("raw_confirmer")

# Prepare the upsert source by joining the latest confirmer data
from pyspark.sql import functions as F

df_upsert = df_predictor.join(
    df_confirmer,
    on="uuid",
    how="left"
).select(
    "uuid",
    "predictions",
    F.coalesce(df_confirmer["code1"], df_predictor["code1"]).alias("code1"),
    F.coalesce(df_confirmer["code2"], df_predictor["code2"]).alias("code2"),
    df_predictor["etl_ingest_ts"],
    F.when(df_confirmer["code1"].isNotNull() | df_confirmer["code2"].isNotNull(), df_confirmer["confirmer_etl_ts"]).otherwise(df_predictor["timestamp_update"]).alias("timestamp_update")
)

# Merge (upsert) into persistent Delta Table
from delta.tables import DeltaTable

deltaTable = DeltaTable.forPath(spark, "/mnt/path/to/data_predictions")
(
    deltaTable.alias("t")
    .merge(
        df_upsert.alias("s"),
        "t.uuid = s.uuid"
    )
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute()
)

Schedule this merge as a Databricks Job every few minutes or hours (frequency depends on latency requirements).
Delta will retain full historical records. Use Delta Time Travel (VERSION AS OF, TIMESTAMP AS OF), schema evolution, and keep the data_predictions table as a true Delta Table, not a view. Pipelines restarts will not destroy historical data.

Additional Tips

Never use streaming joins or self-updating tables inside DLT for this CDC merge. DLT wants a strictly acyclic DAG for tables, and streaming joins typically result in views—not Delta Tables—and lose history.
Consider watermarking and data retention on the raw tables, if storage is a concern.
Late-arriving data is handled: Because all upserts/merges are re-run on new Confirmer data, historical Predictor rows are updated as soon as late Confirmer data arrives.
You can time travel historical states using Delta’s time travel features:

sql

SELECT * FROM data_predictions VERSION AS OF 10
If you want the data_predictions to always reflect only the most recent state per uuid, keep only the latest row per uuid during the upsert.

Illustrated Workflow

Source	DLT Table	Upsert Engine	Gold Table (Delta Table)
Event Hub 1	raw_predictor (Delta)	Batch Notebook/Job	data_predictions (Delta Table)
Event Hub 2	raw_confirmer (Delta)	(merge on uuid)	(all history retained by Delta)

References

[Delta Live Tables Limitations & Best Practices]
[How to use CDC & Upserts with Delta Tables]

Summary

Ingest both streams to separate persistent Delta tables.
Run scheduled upsert jobs to consolidate into a single Delta Table (data_predictions) with history.
Avoid streaming joins and self-updates within DLT—use batch merge logic.
Leverage Delta Lake for historical queries and late data handling.

This approach meets your requirements and works reliably for CDC and late-arriving updates in DLT.