How do I define a streaming table in SDP?

Summary

• In Spark Declarative Pipelines (SDP), you define a streaming table with the @dp.table decorator on a Python function that performs a streaming read, or with CREATE STREAMING TABLE ... AS SELECT ... FROM STREAM ... in SQL.
• The Spark guide describes a streaming table as "a definition of a table and one or more streaming flows written into it" that processes "only new data as it arrives."
• SDP reads any source that Structured Streaming's spark.readStream supports, and the SQL STREAM keyword marks the source as streaming.
• To feed one table from several sources, create the table first and attach one append flow per source.
• On Databricks, Lakeflow Pipelines extend SDP, store streaming tables as Delta tables, and add features such as Auto Loader ingestion and expectations.

How do I define a streaming table in SDP?

In Spark Declarative Pipelines (SDP), you define a streaming table with the @dp.table decorator on a Python function that performs a streaming read, or with CREATE STREAMING TABLE ... AS SELECT ... FROM STREAM ... in SQL. The Spark guide describes a streaming table as "a definition of a table and one or more streaming flows written into it," and says streaming tables support "incremental processing of data, allowing you to process only new data as it arrives." SDP shipped in Apache Spark 4.1.0, and pip install pyspark[pipelines] installs it.

Which approach should I use?

You can define the same kind of dataset in Python or SQL, with one source or several.

Approach	Python	SQL	Use it when
One source	@dp.table on a function that returns a streaming DataFrame	CREATE STREAMING TABLE t AS SELECT * FROM STREAM src	One streaming query feeds the table
Several sources	dp.create_streaming_table("t") plus one @dp.append_flow(target="t") per source	CREATE STREAMING TABLE t; plus one CREATE FLOW f AS INSERT INTO t SELECT * FROM STREAM(src) per source	Several streams append into one table

How do I define one in Python?

Import the module as dp, then decorate a function that returns a streaming DataFrame. SDP supports "all the formats supported by Spark Structured Streaming (spark.readStream)," so a Kafka topic works like a table. The name argument sets the table name:

from pyspark import pipelines as dp
from pyspark.sql import DataFrame

@dp.table
def basic_st() -> DataFrame:
    return spark.readStream.table("samples.nyctaxi.trips")

@dp.table(name="orders_raw")
def ingestion_st() -> DataFrame:
    return (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("subscribe", "orders")
        .load()
    )

Spark 4.2 injects spark into each pipeline file. On Spark 4.1, start the file with spark = SparkSession.active().

How do I define one in SQL?

Use the STREAM keyword to mark the source as streaming:

CREATE STREAMING TABLE basic_st
AS SELECT * FROM STREAM samples.nyctaxi.trips;

How do I feed one table from several sources?

Create the table first, then attach one append flow per source:

from pyspark import pipelines as dp
from pyspark.sql import DataFrame

dp.create_streaming_table("customers_us")

@dp.append_flow(target = "customers_us")
def append_customers_us_west() -> DataFrame:
    return spark.readStream.table("customers_us_west")

@dp.append_flow(target = "customers_us")
def append_customers_us_east() -> DataFrame:
    return spark.readStream.table("customers_us_east")

What does the pipeline need to run it?

The pipeline spec file requires a storage field, "A directory where checkpoints can be stored for streaming tables within the pipeline." With no refresh options, spark-pipelines run performs a default incremental update. spark-pipelines run --full-refresh basic_st clears the data and checkpoints for that dataset and recomputes it from scratch. spark-pipelines dry-run reads and writes no data but catches syntax errors, analysis errors and cyclic dependencies. The Spark 4.1.0 release notes say Spark handles "the execution graph, dependency ordering, parallelism, checkpoints, and retries."
The function must return a pyspark.sql.DataFrame and must never save or write data. SDP evaluates this code more than once during planning and runs.

How does Databricks approach streaming tables?

Lakeflow Pipelines extend SDP and run on the Databricks Runtime. The Databricks docs define a streaming table as "a Delta table with additional support for streaming or incremental data processing," owned and updated by a single pipeline, where each input row is handled only once. They recommend Auto Loader for most ingestion from cloud object storage, and they name Apache Kafka, Azure Event Hubs and Google Pub/Sub as examples of supported message buses. A documented Auto Loader pattern in SQL:

CREATE OR REFRESH STREAMING TABLE customers
AS SELECT * FROM STREAM read_files("/Volumes/my_catalog/retail_org/customers/", format => "csv");

For data quality, Databricks adds expectations, which can fail updates or drop records when they detect invalid ones. In Python, stack @dp.expect("valid_customer_age", "age BETWEEN 0 AND 120") under @dp.table. In SQL, add a CONSTRAINT valid_customer_age EXPECT (age BETWEEN 0 AND 120) clause.
The Databricks docs describe these streaming table behaviors:

Situation	What happens	What to do
You change the table's query	Existing rows aren't updated, only new rows	Trigger a full refresh to requery all previous data
The stream is unbounded and has no watermark	The pipeline can fail from memory pressure	Bound the stream with a watermark
A joined dimension changes	The join isn't recomputed	Use a materialized view if results must always be correct
Records are updated or deleted at the source	Streaming tables are designed for append-only sources	Use AUTO CDC instead of appending (Databricks only)

FAQs

Can a streaming table and a materialized view share a pipeline?

Yes. The guide's example defines orders as a streaming table and customers, customer_orders and daily_orders_by_state as materialized views in one pipeline.

How do I set the table name?

Pass name to the decorator, as in @dp.table(name="orders_raw"). The guide's loop example builds names this way.

Which sources can a streaming table read?

Any format Structured Streaming supports through spark.readStream. The guide lists ingestion from cloud storage and from message buses such as Apache Kafka.