SDP API — Flow types, Destination Types, and Syntax

In the last lecture, we built the vocabulary: four flow types, four destination types, and the pyspark.pipelines (dp) module. Now let's turn that vocabulary into actual syntax — the patterns you'll reach for once you start writing SDP pipelines yourself.

A quick note before we dive in: the code in this lecture is reference syntax, shown to build the mental model for each combination — not a single running notebook. The real end-to-end bronze-to-silver-to-gold pipeline, built and executed against real data, is the hands-on exercise coming up later in this chapter.

SDP flow creation trips up a lot of beginners at first. Once the concepts and combinations click, it becomes simple — so let's build that clarity here, one combination at a time.

The Combination Matrix

Use this table as your reference point for deciding what kind of flow to write, and what's actually allowed:

Flow type	Valid source	Valid target	Decorator / fn	Implicit/Explicit	Status
Append	Auto Loader · Kafka · streaming table (`spark.readStream`)	Streaming table or sink	`@dp.table` / `@dp.append_flow`	Implicit or Explicit	GA
AUTO CDC	Streaming table within the pipeline (`source=` param)	Streaming table only	`dp.create_auto_cdc_flow()`	Explicit only	GA
MV flow	Delta table · streaming table · MV · temp view (`spark.read`)	Materialized view only	`@dp.materialized_view()`	Implicit only	GA
Update	Streaming table producing stateful aggregation (`spark.readStream`)	Sink only (no Delta tables)	`@dp.update_flow`	Explicit only	Preview

Seven combinations come out of this matrix, and we'll walk through each one.

Append Flow: Three Combinations

Append flow reads from a streaming source (spark.readStream) and writes to a streaming table or a sink. It's GA, and it can be created two ways: implicit (you just define the target table, and SDP creates the flow for you) or explicit (you call dp.append_flow yourself).

Combination 1a — Streaming table, implicit form

This is the bronze pattern — the single most common form you'll write. @dp.table declares the streaming table and the append flow into it, in one step. That's what "implicit" means.


python
from pyspark import pipelines as dp

@dp.table(name="bronze_orders_cdc",
          comment="Raw CDC events — append only")
@dp.expect_or_drop("ts_ms_not_null", "ts_ms IS NOT NULL")
def bronze_orders_cdc():
    return (
        spark.readStream                       # ← must be readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "csv")
            .option("cloudFiles.schemaLocation", "/path/to/schema")
            .load("/path/to/source/")
    )

Notice what's missing compared to the raw Auto Loader script from earlier in the course: no checkpoint location, no explicit write stream call, no .awaitTermination(). You define the target table, the data quality rule, and where to read from — SDP handles the rest.

One convention worth calling out: the function name doesn't have to match the table name, but it's good practice to keep them the same. If you omit name= in the @dp.table(...) annotation entirely, SDP will use the function's name as the table name instead.

Combination 1b — Streaming table, explicit form (multiple sources)

Implicit form breaks down the moment you have more than one source feeding the same target. Say you have a bronze_events table that needs to ingest from two different landing folders — orders and returns — because they represent the same kind of business event. Implicit form only lets you wire one source to one target. For multiple sources into a shared target, you need explicit form.

Explicit form is always two steps: create the table, then define the flow(s) into it.


python
from pyspark import pipelines as dp

# Step 1: declare the target streaming table separately
dp.create_streaming_table(
    name="bronze_events",
    comment="Unified events table — multiple sources"
)

# Step 2a: first append flow into it
@dp.append_flow(target="bronze_events", name="orders_flow")
def orders_flow():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("/Volumes/dev/dbx_course/landing/orders/")
    )

# Step 2b: second append flow into the same table
@dp.append_flow(target="bronze_events", name="returns_flow")
def returns_flow():
    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("/Volumes/dev/dbx_course/landing/returns/")
    )

Both flows append into the same bronze_events streaming table, and SDP wires them together automatically — no manual dependency management needed.

Combination 1c — Sink, explicit form

Append flow isn't limited to writing into Unity Catalog tables — it can also write to an external sink, like Kafka. This always uses explicit form, since implicit form exists purely for simplicity, not this kind of flexibility. Same two-step pattern: create the sink, then define the flow into it.


python
from pyspark import pipelines as dp

# Step 1: declare the sink
dp.create_sink(
    name    = "kafka_orders_sink",
    format  = "kafka",
    options = {
        "kafka.bootstrap.servers": "<broker-address>",
        "topic": "processed-orders",
    }
)

# Step 2: append flow into the sink
@dp.append_flow(target="kafka_orders_sink", name="orders_to_kafka")
def orders_to_kafka():
    return (
        spark.readStream.table("silver_orders")
            .filter("__END_AT IS NULL")
            .select("order_id", "customer_id", "amount", "status")
    )

AUTO CDC Flow: The Silver Pattern

You can use append flow for silver-layer logic — reading from bronze, applying transformations, writing to silver. But if you're implementing a proper CDC pattern with SCD Type 1 or Type 2, AUTO CDC is the dedicated flow built for exactly that. It's always explicit — there's no implicit form.

Combination 2 — Streaming table with AUTO CDC flow

Same two-step shape as explicit append flow: declare the target table first, then define the flow.


python
from pyspark import pipelines as dp

# Step 1: declare the target streaming table
dp.create_streaming_table(
    name="silver_orders",
    comment="SCD Type 2 order history",
    expect_all_or_drop={
        "valid_order_id": "order_id IS NOT NULL",
        "valid_amount":   "amount > 0",
    }
)

# Step 2: define the AUTO CDC flow into it
dp.create_auto_cdc_flow(
    target        = "silver_orders",
    source        = "bronze_orders_cdc",
    keys          = ["order_id"],
    sequence_by   = "ts_ms",
    apply_as_deletes   = "op = 'd' OR __deleted = 'true'",
    stored_as_scd_type = "2",
    except_column_list = ["op", "ts_ms", "__deleted",
                           "_ingest_timestamp", "_source_file"],
)

A few things worth calling out in the parameters:

sequence_by — the column AUTO CDC uses to determine which record is newest, so it can correctly handle records that arrive out of order.
apply_as_deletes — the rule for identifying which incoming records represent a delete. Inserts and updates are handled automatically through a merge (matched primary key = update, unmatched = insert); deletes need an explicit rule because there's no other way to distinguish them.
stored_as_scd_type — "1" or "2", depending on whether you want history preserved.
except_column_list — columns from the source you don't want carried into the target (CDC metadata columns like op, ts_ms, __deleted typically don't belong in the business-facing silver table).

Notice one difference from the implicit @dp.table pattern: when you create the table explicitly with dp.create_streaming_table, data quality rules are passed as a parameter (expect_all_or_drop={...}) right in the table creation call, rather than as separate decorators.

MV Flow: The Gold Pattern

Materialized view flow is a batch flow — you write full-table transformation logic without worrying about incremental processing at all, and SDP's incremental refresh engine figures out what changed.

Combination 3 — Materialized view, implicit only

MV flow has no explicit form. @dp.materialized_view is the only way to declare it — and that's by design. A gold-layer aggregation has to come from one single query; it wouldn't make sense to let multiple flows write into the same aggregated result the way multiple append flows can share a bronze table. Forcing implicit-only prevents that mistake entirely.


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

@dp.materialized_view(
    name="gold_daily_revenue",
    comment="Daily revenue by customer tier"
)
def gold_daily_revenue():
    orders = (
        spark.read.table("silver_orders")        # ← must be spark.read, never readStream
            .filter(F.col("__END_AT").isNull())  # SCD Type 2 current-record filter
            .filter(F.col("status") == "completed")
    )
    customers = (
        spark.read.table("silver_customers")
            .filter(F.col("__END_AT").isNull())
            .select("customer_id", "customer_tier")
    )
    return (
        orders.join(customers, on="customer_id", how="left")
            .groupBy(
                F.to_date(F.col("order_date")).alias("order_date"),
                F.col("customer_tier")
            )
            .agg(
                F.sum("amount").alias("total_revenue"),
                F.count("order_id").alias("order_count"),
            )
    )

Here's the part that genuinely surprises people coming from a data warehousing background: this code reads the entire silver_orders and silver_customers tables, joins them, and aggregates — with zero incremental logic written anywhere. By the letter of the code, this should be a full recompute every single run.

It isn't. SDP's materialized view engine figures out, most of the time, how to refresh this incrementally behind the scenes — processing only what changed rather than reprocessing everything. Occasionally the underlying logic doesn't meet the criteria for incremental refresh, and SDP falls back to a full refresh instead — but the common case is incremental, even though you never wrote incremental logic. This is genuinely one of the most powerful parts of SDP.

One important warning baked into the naming here: if you're coming from a relational database or data warehouse background, "materialized view" there usually just means a query result permanently stored as a table. In SDP, that's still true, but there's always a flow behind an SDP materialized view — not a plain query. Even Databricks SQL's own CREATE MATERIALIZED VIEW syntax creates an MV flow behind the scenes. The flow is what captures the logic for how the data got there.

Update Flow: Streaming Aggregations to External Systems

If append flow can already write to a sink (combination 1c), why does update flow exist? Because append flow is append-only — it can send new rows to a Kafka topic, but it can't send updated values. Update flow exists specifically for streaming aggregations or running summaries that need to be refreshed as new data arrives — the same kind of result a gold-layer materialized view produces, but sent to an external system instead of a Unity Catalog table.

Combination 4 — Sink with update flow (streaming aggregation)

Explicit only, same two-step shape as before.


python
from pyspark import pipelines as dp
from pyspark.sql.functions import col

# Step 1: declare the sink
dp.create_sink(
    name    = "kafka_order_counts_sink",
    format  = "kafka",
    options = {
        "kafka.bootstrap.servers": "<broker-address>",
        "topic": "order-counts-by-status",
    }
)

# Step 2: update flow — stateful aggregation, emits only changed rows per batch
@dp.update_flow(target="kafka_order_counts_sink", name="order_counts_flow")
def order_counts_flow():
    return (
        spark.readStream.table("bronze_orders_cdc")
            .groupBy(col("status"))
            .count()                              # ← stateful aggregation, no watermark needed
    )

The key phrase in the comment above the flow: this is a stateful aggregation, and unlike raw Structured Streaming aggregations, you don't need to reason about watermarks yourself.

Temporary View: Shared Intermediate Logic

Sometimes a silver-to-gold transformation can't be a single step — or several gold tables need to share the same filtering logic, and you don't want that logic duplicated across every MV flow that uses it. This is exactly what temporary views solve.

Combination 5 — Temporary view

Also always two steps: define the temp view once, then reference it downstream by name, just like any other table.


python
from pyspark import pipelines as dp

@dp.temporary_view(name="current_orders")
def current_orders():
    return (
        spark.read.table("silver_orders")
            .filter("__END_AT IS NULL")
            .filter("status = 'completed'")
    )

Reference it downstream by name, just like any other table:


python
@dp.materialized_view(name="gold_revenue_by_tier")
def gold_revenue_by_tier():
    return spark.read.table("current_orders").groupBy("customer_tier").agg(...)

@dp.materialized_view(name="gold_revenue_by_region")
def gold_revenue_by_region():
    return spark.read.table("current_orders").groupBy("region").agg(...)

The filter logic runs once, via the temp view, and both gold tables share it — nothing is duplicated between them. Remember: a temp view is scoped to the pipeline's lifetime only. It's never published to Unity Catalog, and it can't be queried from outside the pipeline.

Putting It Together: The Decision Framework

With seven combinations on the table, here's the actual decision path to follow:

Start here: does your output need to live in Unity Catalog, or does it go to an external system?

Unity Catalog (managed table): use a streaming table or a materialized view.
External system (Kafka, Event Hubs, external Delta): use a sink.

If streaming table:

New rows arriving incrementally, from one source → @dp.table (implicit append flow)
New rows arriving from multiple sources → dp.create_streaming_table() + @dp.append_flow (explicit)
CDC events (inserts/updates/deletes) → dp.create_streaming_table() + dp.create_auto_cdc_flow()

If materialized view:

Always → @dp.materialized_view + spark.read (never spark.read_stream — that will throw an error)

If sink:

Forward processed records to an external system → dp.create_sink() + @dp.append_flow
Stream a running aggregation to an external system → dp.create_sink() + @dp.update_flow

Summary

Combination	Pattern	Layer	Implicit/Explicit
1a	`@dp.table` reading `readStream`	Bronze	Implicit
1b	`create_streaming_table` + multiple `@dp.append_flow`	Bronze (multi-source)	Explicit
1c	`create_sink` + `@dp.append_flow`	Bronze/Silver → external	Explicit
2	`create_streaming_table` + `create_auto_cdc_flow`	Silver (CDC/SCD)	Explicit only
3	`@dp.materialized_view` reading `spark.read`	Gold	Implicit only
4	`create_sink` + `@dp.update_flow`	Gold-style aggregation → external	Explicit only
5	`@dp.temporary_view`	Shared intermediate logic	Implicit

Next lecture, we put all of this into practice and build a complete end-to-end pipeline — bronze to silver to gold — so you can see the mechanics working together on real data.

See you again. Keep learning, and keep growing!