DLT Pipeline - Overwrite except for one Append table

[Melia-Taylour]

Hi all,

I am a junior engineer and I am working on a use case involving DLT pipelines that read from a csv source at regular intervals and populate tables based upon processing of this. In most cases, I'd like the table contents to be overwritten with the new data, however there is one table where I need to append the new data instead to create a historical view.

To put the situation in pseudocode:

import pipelines as dp

dp.table(name = "Table1") # Overwrite
def table1():
    return spark.read.csv(path_to_my_csv).do_some_simple_processing()

dp.table(name = "Table2") # Need to have this append rather than overwrite
def table2():
    return spark.read.table("Table1")

I've looked and I don't think this is covered by any of the option flags in the table decorator. I've also seen streaming_table and append_flow as potentially recommended, however they seem to require the source itself to be streaming tables which I don't know if this is possible for my use case. Can I ask what the recommended way to achieve this outcome with declarative pipelines would be?

Thanks!

[Satyasai]

Hi @Melia-Taylour 

In DLT/SDP, a streaming_table does not require your raw source (the CSV) to be a streaming source. When you read a standard Delta table with spark.readStream, Delta Lake interprets the table's transaction log as a stream of commits to the table.

Each time you overwrite your CSV with updated data and Table1 is recomputed/overwritten, the new rows will show up as a new commit in Table1 for Table2 to read from.
Recommended Implementation

Python
import dlt
from pyspark.sql import functions as F
# 1. Batch Table: Overwrites on every pipeline run
@dlt.table(
name="Table1",
comment="Batch table refreshed with the latest CSV snapshot"
)
def table1():
# Regular batch read from CSV
return (
spark.read
.option("header", "true")
.csv("/path/to/source.csv")
.withColumn("ingested_at", F.current_timestamp())
)
# 2. Historical Table: Incrementally appends new batch commits
@dlt.table(
name="Table2",
comment="Append-only historical archive of Table1 snapshot runs"
)
def table2():
# spark.readStream turns Table1's transaction log into a stream,
# appending new incoming batches without overwriting existing history.
return spark.readStream.table("LIVE.Table1")

How It Works Under the Hood
Table1 (Batch): This is a regular materialized view. On every pipeline run, it will overwrite itself with the latest CSV snapshot.
Table2 (Streaming Table): This is a streaming query that uses the streaming reader to read from Table1's transaction log. By using spark.readStream.table("LIVE.Table1"), any new commits to Table1 (i.e. new CSV snapshots) will show up as new records in Table2.
Important consideration: Overwrites vs Appends in Table1
If Table1 does a full overwrite on every run (i.e. replaces existing keys with updated values) and Delta logs record those as row deletions, you may need to set the skipChangeCommits option when reading Table1 as a stream:
Python
@dlt.table(name="Table2")
def table2():
return (
spark.readStream
.option("skipChangeCommits", "true")
.table("LIVE.Table1")
)
Alternative: Auto Loader directly for History
If your CSV source contains sequentially named files (data_2026_09_01.csv, data_2026_09_02.csv, etc.) that get periodically ingested, you can also use Auto Loader (spark.readStream.format("cloudFiles")) to directly load Table2's history from storage. This approach will append every CSV file as a new record in Table2.

[data_pulse]

@Melia-Taylour 

For Table1 @dp.table approach is fine if you want it to represent the latest/current state on each pipeline update.

For Table2, use create_auto_cdc_from_snapshot_flow with scd_type=2

dp.create_streaming_table("Table2")
dp.create_auto_cdc_from_snapshot_flow(
    target="Table2",
    source="Table1",
    keys=["id"],
    stored_as_scd_type=2
)

Table1 can represent the latest snapshot on each run and create_auto_cdc_from_snapshot_flow() compares that snapshot with the previous one. It updates Table2 only when a key is new, changed or removed. With SCD Type 2, changed rows create new history versions using __START_AT and __END_AT. 

If the new snapshot is identical to the previous one, no new history rows are created. So you get change based history without needing a streaming source or custom append logic.

Let me know if that answers your question. If not, post an example on how the history has to be retained.

[AbhilashNagilla]

If history means changed versions by key, define Table1 as a batch materialized view and feed it to AUTO CDC FROM SNAPSHOT; its source can be a table or view, so the CSV doesn't need streaming semantics (Python datasets, snapshot API).

from pyspark import pipelines as dp

dp.create_streaming_table("Table2")
dp.create_auto_cdc_from_snapshot_flow(
    target="Table2",
    source="Table1",
    keys=["id"],
    stored_as_scd_type=2,
)

Replace id with the column or columns that uniquely identify a source row; SCD Type 2 adds a version when values for an existing key change (snapshot API, CDC examples). This API requires serverless Lakeflow pipelines or the Pro or Advanced edition (CDC requirements).

The table/view form reads one snapshot per update; if snapshots can accumulate between updates, use the documented version-function source to process them in order (snapshot examples).

append_flow requires streaming input unless once=True, which runs batch input once; skipChangeCommits ignores modifying commits, so it won't archive Table1 overwrites (append flow, Delta streaming).

Use regular updates because a full refresh clears streaming-table data and flow checkpoints before rebuilding from available source data (update semantics).

If history means every delivery row, land each CSV as a new file and ingest Table2 directly with Auto Loader (Auto Loader).