Streaming Tables, MVs, Flows & Sinks

The one-liner

The four objects — click each

▸ streaming table · materialized view · flow · sink

1 · Streaming table or materialized view? — click to reveal

Pick the right object for each scenario:

Python (current API) & SQL

from pyspark import pipelines as dp
@dp.table                       # streaming table — incremental ingest
def bronze_events():
    return spark.readStream.format("cloudFiles").option("cloudFiles.format","json").load(src)

@dp.materialized_view           # MV — always-correct join to current dim
def user_activity():
    return spark.read.table("bronze_events").join(spark.read.table("dim_users"), "user_id")

CREATE OR REFRESH STREAMING TABLE bronze_events
AS SELECT * FROM STREAM read_files('/Volumes/main/raw/events', format => 'json');
CREATE OR REFRESH MATERIALIZED VIEW user_activity
AS SELECT e.*, u.segment FROM bronze_events e JOIN dim_users u USING (user_id);

2 · Flows — fan many sources into one ST

Create the streaming table explicitly, then attach append flows — each processes incrementally and independently, feeding one ST (cleaner than a giant UNION that reprocesses everything).

dp.create_streaming_table("all_events")        # target ST (no query of its own)

@dp.append_flow(target="all_events")          # flow 1: US
def us_events():  return spark.readStream.table("bronze_events_us")

@dp.append_flow(target="all_events")          # flow 2: EU — same target
def eu_events():  return spark.readStream.table("bronze_events_eu")

3 · Sinks — write out to Kafka / Event Hubs / Delta

Create a sink with create_sink, then point an append flow at it. Sinks are Python-only and accept streaming queries only.

dp.create_sink(
    name="events_kafka",
    format="kafka",                            # also: "delta", Event Hubs (via Kafka), custom
    options={"kafka.bootstrap.servers": "host:9092", "topic": "enriched_events"})

@dp.append_flow(target="events_kafka")        # write the stream to the sink
def publish_events():
    return spark.readStream.table("user_activity").selectExpr(
        "CAST(user_id AS STRING) AS key", "to_json(struct(*)) AS value")

4 · Stateful streams need watermarks

Streaming aggregations/joins keep state. Without a watermark bounding how late data can arrive, that state grows unbounded → OOM.

@dp.table
def events_per_min():
    from pyspark.sql.functions import window, col
    return (spark.readStream.table("bronze_events")
              .withWatermark("event_ts", "10 minutes")   # bound late data → bounded state
              .groupBy(window(col("event_ts"), "1 minute")).count())

Why it matters

• ST vs MV is the #1 SDP interview question — pick wrong and you reprocess everything (slow) or serve stale joins (wrong).
• Flows + sinks show SDP is more than two table types — fan multiple inputs into one ST and write out to Kafka.

Uses, edge cases & limitations

• Uses: STs for bronze/silver ingest; MVs for gold aggregates & dimension-joined tables; append flows to fan many sources into one ST; sinks to push results out.
• Edge cases: STs are append-mostly — schema changes/reprocessing often need a full refresh; unbounded streaming aggs/joins need watermarks or state grows unbounded; an ST is owned by a single pipeline.
• Limitation: MV refresh has latency/cost; STs can be "fast-but-wrong" if a dimension changed after a row was processed (so joins often belong in MVs). Sinks: Python-only, streaming-only.

Common gotchas

• ❌ Streaming table for a dimension join that must stay current → use an MV.
• ❌ Materialized view for high-volume append ingest → use a streaming table.
• ❌ Trying to define a sink in SQL — sinks are Python-only, streaming-only.
• ❌ Forgetting watermarks on unbounded streams → unbounded state / OOM.
• ❌ Writing to one streaming table from two pipelines → conflicts (single owner).