Databricks Community

Melia-Taylour · 3 weeks ago

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 · 3 weeks ago

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 · 3 weeks ago

@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 · 3 weeks ago

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).

anuj_lathi · 2 hours ago

Short version: your Table1 is really a materialized view (batch read, fully recomputed each update), and Table2 needs to be a streaming table fed by an incremental source, not by Table1. The fix is to change how you ingest the CSV for the history table, not to find a hidden flag on the decorator.

What's going on

@DP.table defines streaming tables and expects a function that returns a streaming DataFrame. @DP.materialized_view is the right decorator for batch logic that's recomputed each run [1]. The older dlt module used @table for both, which is why your pseudocode looks like it should work [1][2].
Table1 is the "overwrite" case and works as a materialized view. Don't try to stream from it. Its contents get rewritten every update, and a streaming table needs an append-only source.
You're right that append_flow needs a streaming query. The one exception is once=True, which accepts a batch DataFrame, but that flow only runs once (and again on a full refresh) [3]. It's a backfill tool, not a recurring "append every run" mechanism.

Recommended pattern: ingest the CSVs incrementally with Auto Loader

Your CSV source doesn't have to be a streaming table. Auto Loader turns a folder of CSV files into a streaming read, and only new files are processed on each update, so the target table naturally accumulates history. Keep the shared transformation in a plain function so both tables use the same logic [4]:

from pyspark import pipelines as dp
from pyspark.sql import functions as F

def do_some_simple_processing(df):
    ...  # your shared transformation logic

# Overwrite semantics: recomputed from the source every update
@dp.materialized_view(name="table1")
def table1():
    return do_some_simple_processing(spark.read.csv(path_to_my_csv, header=True))

# Append semantics: only new files are processed, rows accumulate
@dp.table(name="table2")
def table2():
    df = (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .load(path_to_my_csv_folder)
    )
    return (
        do_some_simple_processing(df)
        .withColumn("source_file", F.col("_metadata.file_path"))
        .withColumn("ingested_at", F.current_timestamp())
    )

Things to watch:

Land each delivery as a new file (new filename or dated subfolder) rather than overwriting the same path. Auto Loader tracks files it has already seen, so a file rewritten in place won't be picked up as new by default.
Adding source_file and ingested_at columns gives you a proper historical view of which load each row came from.
Don't use .write, saveAsTable, toTable, etc. inside a dataset function to force an append. Pipeline dataset code should only return a DataFrame [5].

If you really need history of Table1's output

If what you want is "keep a record of how the processed snapshot changed over time" and the source is a periodic full snapshot, look at create_auto_cdc_from_snapshot_flow. It's built for deriving change history (including SCD2) from successive snapshots [5]. The target is declared with create_streaming_table() [6].

Recommendation

For most "overwrite these, append that" setups, use the materialized view for Table1 and an Auto Loader streaming table for Table2, both reading from the same CSV landing location. If you can't change how the files arrive (same filename overwritten each time), tell me how the CSV is delivered and I can suggest the cleanest way to handle that.