topic Re: Databricks DLT Joins: Streaming table join with Delta table is reading 2 Billion records per bat in Data Engineering

Databricks DLT Joins: Streaming table join with Delta table is reading 2 Billion records per batch

gudurusreddy99 — Thu, 23 Oct 2025 07:57:06 GMT

Databricks DLT Joins: Streaming table join with Delta table is reading 2 Billion records from Delta Table for each and every Micro batch.

How to overcome this issue to not to read 2 Billion records for every micro batch.

Your suggestions and feedback will be highly appreciated.

Thanks

Srini

Re: Databricks DLT Joins: Streaming table join with Delta table is reading 2 Billion records per bat

ManojkMohan — Thu, 23 Oct 2025 09:51:04 GMT

@gudurusreddy99

Root Cause
When performing stream-static joins, which means joining a streaming source with a static Delta table, Spark will scan the entire Delta table for each micro-batch if the static table is not filtered or incrementally joined.

https://www.linkedin.com/pulse/overcoming-databricks-dlt-streaming-table-join-limitations-roger-ding-kw0ie/

Solution Thinking:

1. Incremental Join Logic
Build your pipeline to join only new or changed data in the Delta table since the last batch with the current micro-batch from the streaming source.

Use watermarks and incremental filter conditions to limit joined records to the relevant time window or ID range. This will reduce the scan of the static Delta table.

https://community.databricks.com/t5/data-engineering/incremental-join-transformation-using-delta-live-tables/td-p/64725

2. Partitioning and Z-Ordering
Partition the static Delta table based on the join key, such as customer_id. This will help optimize data skipping during joins.

Apply Z-Ordering on the main join keys to improve clustering and further decrease file I/O and scan overhead.

3. Broadcast Joins (for Small Reference Tables)
If the Delta table is relatively small, enable broadcast joins in your query.

https://community.databricks.com/t5/technical-blog/top-5-tips-to-build-delta-live-tables-dlt-pipelines-optimally/ba-p/83871

4. Use Change Data Capture (CDC) or Materialized Views
For Delta tables that change frequently or grow over time, implement CDC to process only updated rows in each batch.

Alternatively, you can materialize incremental results in intermediate tables and join streaming input with only the daily or hourly chunk, not the entire history.

https://stackoverflow.com/questions/75094547/creating-a-delta-live-table-for-reading-incremental-data

Next Steps:
Examine the partition structure of your current Delta table and make improvements as needed.

Add time or ID-based filters in your static or lookup table scans whenever possible.

If applicable, refactor your DLT pipeline to maintain gold or silver incremental tables as pre-filtered join sources.

Re: Databricks DLT Joins: Streaming table join with Delta table is reading 2 Billion records per bat

ManojkMohan — Fri, 24 Oct 2025 10:45:05 GMT

@gudurusreddy99 Any update here , did you try the above solutions ?

Re: Databricks DLT Joins: Streaming table join with Delta table is reading 2 Billion records per bat

gudurusreddy99 — Sun, 02 Nov 2025 16:09:18 GMT

Thanks for your response @ManojkMohan

Re: Databricks DLT Joins: Streaming table join with Delta table is reading 2 Billion records per bat

gudurusreddy99 — Sun, 02 Nov 2025 16:18:17 GMT

Requirement:

I have Kafka streaming Pixels data and I want to check incoming Pixels key with one of the existing Delta table which holds 2 Billion+ records and this table keep getting appended by another Databricks process.

Issue: At the moment, my solution is working to check the matching records which are older than 10min. My process should check all the up to date records from pixel_tracking_data delta table which holds data from more than 2 years and over 2 Billion records with low latency.

The current process is taking 7-8 seconds to match records from 2 Billion records and Delta table is having liquid partitioning.

Please suggest any recommendations.

The code being used:

@dlt.view

def lookup_keys():

return (

dlt.read_stream("pixels_db_df_dlt")

.where(F.col('ExecException').isNull())

.select(

F.upper(F.col("TrackingTag")).alias("BTag"),

F.upper(F.col("SubscriberGuid")).alias("SubscriberGuid")

)

.distinct()

)

@dlt.table(name="pixel_subset")

def pixel_subset():

keys = dlt.read_stream("lookup_keys")

# # Force Delta metadata refresh so Spark sees new files quickly

# spark.catalog.refreshTable(f"operational_{env}.silver_data.pixel_tracking_data")

base = (spark.read.table(f"operational_{env}.silver_data.pixel_tracking_data")

.select(

F.col("OperatorId").alias("SubscriberGuid"),

F.col("BTag"),

F.col("PublishPointID"),

F.col("AffiliateID"),

F.col("extraParameters"),

F.col("AdID").alias("AdId"),

F.col("MarketingSourceGroupID").alias("SourceGroupID")

))

return (

keys.join(base,["SubscriberGuid", "BTag"], "inner")

)

@dlt.table(name="lookup_df")

def lookup_df_combined():

logger.info(str(dt.now()) + " - Started reading data for lookup_df_combined")

pixels_df = (dlt.read_stream("pixels_db_df_dlt")

.where(F.col('ExecException').isNull())

.drop(F.col('ExecException'))

)

matched_btags = spark.read.table(f"operational_{env}.silver_data.pixel_subset")

# Step 4: Perform final join (combination of both tables)

result = (

pixels_df.alias('curr')

.join(

matched_btags.alias('lookup'),

(F.col("curr.SubscriberGuid") == F.col("lookup.SubscriberGuid")) & (F.col("curr.TrackingTag") == F.col("lookup.BTag")) ,'left'

)

.withColumn('retry', F.lit(0))

.withColumn(

'resolved',

F.when(F.col('lookup.BTag').isNotNull(), F.lit(True)).otherwise(False)

)

.select(

F.col('curr.*'),

F.col('lookup.PublishPointID').alias('PublishPointID'),

F.col('lookup.AffiliateID').alias('AffiliateID'),

# F.col('lookup.AdLogActionID').alias('AdLogActionID'),

F.col('lookup.extraParameters').alias('extraParameters'),

F.col('lookup.SourceGroupID').alias('SourceGroupID'),

F.col('lookup.AdID').alias('AdId'),

F.col('retry'),

F.col('resolved')

).withColumn("commit_timestamp",F.expr("current_timestamp()"))

.withColumn("loaded_at", F.expr("current_timestamp()"))

)

logger.info(str(dt.now()) + " - Finished processing lookup_df_combined")

return result