Module 9: Lakeflow Pipelines

• A pipeline = a set of datasets (tables, views) plus the queries that fill them
• It handles order (from dependencies), parallelism, retries, checkpoints and incremental processing
• Writes managed tables in Unity Catalog. Runs on serverless compute by default

• Flow = one query that writes into a table. A table has a default flow
• Sink = a streaming target outside the pipeline (Delta, Kafka, Event Hubs)
• Each dataset is defined once in a workspace. Exception: streaming tables can take several append flows (topic 7)

Create it:

1. New → ETL pipeline
2. Name it dbx_orders. Default catalog: your catalog. Default schema: dbx_pipeline
3. Choose Python (or SQL) for the starter file in transformations
4. Leave compute as serverless (the default)

• Editor parts: file browser, code editor, pipeline graph, data preview, issues, event log
• Dry run checks code and names without writing data
• Run file updates only the tables in the open file. Run pipeline updates everything
• Run the whole pipeline before you call it done: Run file can hide errors in other files

Put this in transformations/orders.py:

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

@dp.table(comment="Raw orders as received")
def orders_bronze():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("pathGlobFilter", "*.json")
        .load(spark.conf.get("course.landing"))
        .withColumn("_ingested_at", F.current_timestamp())
    )

SQL equivalent in transformations/orders.sql:

CREATE OR REFRESH STREAMING TABLE orders_bronze
COMMENT 'Raw orders as received'
AS SELECT *, current_timestamp() AS _ingested_at
FROM STREAM read_files('${course.landing}', format => 'json');

• The function returns a streaming DataFrame. Don't call collect(), count(), saveAsTable() or toTable() inside it
• The table name is the function name (or name=)
• Schema and checkpoints are managed for you: no schemaLocation or checkpointLocation
• The path comes from a pipeline parameter, course.landing (topic 11)
• @dp.table(...) options: comment, cluster_by, partition_cols, schema, table_properties

@dp.materialized_view(comment="Revenue per day, valid completed or shipped orders")
def gold_daily_revenue():
    o = spark.read.table("orders_silver").filter("status IN ('completed', 'shipped')")
    p = spark.read.table("dbx_course.products")
    return (
        o.join(p, "product_id")
         .groupBy("order_date")
         .agg(F.count("*").alias("orders"),
              F.round(F.sum(F.col("quantity") * F.col("unit_price")), 2).alias("revenue"))
    )

CREATE MATERIALIZED VIEW gold_daily_revenue AS
SELECT order_date, COUNT(*) AS orders, ROUND(SUM(quantity * unit_price), 2) AS revenue
FROM orders_silver o JOIN dbx_course.products p USING (product_id)
WHERE status IN ('completed', 'shipped')
GROUP BY order_date;

• Written as a batch query: spark.read.table, not readStream
• Each update brings it in line with the sources, incrementally if Databricks can
• Table names without a schema resolve to the pipeline's default catalog and schema
• Other tables need a schema (dbx_course.products)
• refresh_policy: auto (default), incremental, incremental_strict, full

• You write what: a dataset and the query that defines it
• You don't write how: order, checkpoints, retries, upserts
• Read another dataset by name. That one line creates a dependency edge

@dp.table
def orders_silver():
    return (
        spark.readStream.table("orders_bronze")          # reads the table above
        .withColumn("status", F.lower(F.trim("status")))
        .withColumn("order_date", F.to_date("order_ts"))
        .dropDuplicates(["order_id"])
    )

• Streaming reads use spark.readStream.table(...). Batch reads use spark.read.table(...)
• Streaming dropDuplicates without a watermark keeps state for every key. Fine at this size, costly at scale (Module 11)
• Use views for steps you don't want to publish (below)
• No pivot() (needs the full data up front). Use conditional aggregation

@dp.view                                     # not stored, only inside the pipeline
def valid_orders():
    return spark.readStream.table("orders_bronze").filter("quantity > 0")

@dp.table(private=True)                      # stored, but not published to the catalog
def orders_enriched():
    return spark.readStream.table("valid_orders")

Several sources into one table:

dp.create_streaming_table("all_orders")

@dp.append_flow(target="all_orders", name="web")
def from_web():
    return spark.readStream.table("web_orders")

@dp.append_flow(target="all_orders", name="shop")
def from_shop():
    return spark.readStream.table("shop_orders")

@dp.append_flow(target="all_orders", name="history_backfill", once=True)
def backfill():
    return spark.read.table("old_orders")        # batch, loaded one time

• A flow reads a source, applies logic, writes to a target
• Append flows add new sources without a UNION
• once=True loads history one time (batch source). It reruns only on a full refresh
• The pipeline runs independent flows in parallel

@dp.table
@dp.expect("has_customer", "customer_id IS NOT NULL")             # keep row, log it
@dp.expect_or_drop("positive_qty", "quantity > 0")                # drop row, log it
@dp.expect_or_fail("has_order_id", "order_id IS NOT NULL")        # stop the update
def orders_clean():
    return spark.readStream.table("orders_silver")

CREATE OR REFRESH STREAMING TABLE orders_clean (
  CONSTRAINT has_customer EXPECT (customer_id IS NOT NULL),
  CONSTRAINT positive_qty EXPECT (quantity > 0) ON VIOLATION DROP ROW,
  CONSTRAINT has_order_id EXPECT (order_id IS NOT NULL) ON VIOLATION FAIL UPDATE
) AS SELECT * FROM STREAM orders_silver;

rules = {"has_customer": "customer_id IS NOT NULL", "has_ts": "order_ts IS NOT NULL"}

@dp.table
@dp.expect_all_or_drop(rules)
def orders_checked():
    return spark.readStream.table("orders_silver")

• Decorators go after @dp.table and above the function
• Descriptions must be unique per dataset
• Counts of passed, failed and dropped rows go to the event log (topic 12)
• Dropped rows vanish. To keep them, route failures to a quarantine table with the inverse condition

@dp.view
def customer_changes():
    return spark.readStream.table("dbx_course.customers_feed")

dp.create_streaming_table("customers_current")

dp.create_auto_cdc_flow(
    target="customers_current",
    source="customer_changes",
    keys=["customer_id"],
    sequence_by="seq",
    apply_as_deletes=F.expr("operation = 'DELETE'"),
    except_column_list=["operation", "seq"],
    stored_as_scd_type=1,
)

CREATE OR REFRESH STREAMING TABLE customers_history;

CREATE FLOW customers_cdc AS AUTO CDC INTO customers_history
FROM STREAM(dbx_course.customers_feed)
KEYS (customer_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY seq
COLUMNS * EXCEPT (operation, seq)
STORED AS SCD TYPE 2;

• Add changes by inserting rows into customers_feed with a higher seq, then update the pipeline
• sequence_by handles late and out-of-order events: the highest sequence wins
• The source must be a stream. The target is a streaming table created first
• This replaces the hand-written SCD 2 MERGE from Module 8.10
• apply_changes() is the older name for create_auto_cdc_flow()

• Normal refresh: streaming tables process new records. Materialized views use an incremental refresh when cheaper
• Full refresh: streaming tables clear data and checkpoints and reprocess the source. Source files that no longer exist are lost
• Incremental refresh of materialized views needs serverless compute
• Techniques seen in the event log: ROW_BASED, GROUP_AGGREGATE, WINDOW_FUNCTION, APPEND_ONLY, PARTITION_OVERWRITE, FULL_RECOMPUTE, NO_OP
• Float SUM/AVG can force full recompute: cast to DECIMAL
• WITH RECURSIVE and non-deterministic functions (except time functions in WHERE) fall back to full

-- which technique was used?
SELECT timestamp, message
FROM event_log(TABLE(<catalog>.dbx_pipeline.gold_daily_revenue))
WHERE event_type = 'planning_information'
ORDER BY timestamp DESC

Set it in pipeline Settings → Configuration:

Read it:

min_qty = int(spark.conf.get("course.min_qty"))      # always a string

@dp.table
def orders_valid():
    return spark.readStream.table("orders_bronze").filter(F.col("quantity") >= min_qty)

SELECT * FROM STREAM orders_bronze WHERE quantity >= ${course.min_qty};

• Configuration keys are read with spark.conf.get("key") in Python and ${key} in SQL
• Values are always strings: cast them
• Use dotted names (course.min_qty) so they don't clash with Spark settings
• Newer pipeline parameters are set per update and used with :name in SQL only, e.g. USE CATALOG IDENTIFIER(:source_catalog)
• Per-environment values (dev, prod) belong here: same code, different configuration

• The pipeline page shows each update, its graph, row counts and data quality numbers
• Click a table in the graph for flow metrics, expectation results and the schema
• The event log is a Delta table with every event of every update

-- by pipeline id (copy it from the pipeline settings)
SELECT timestamp, event_type, message
FROM event_log('<pipeline-id>')
ORDER BY timestamp DESC
LIMIT 20

• Expectation results are in details:flow_progress.data_quality.expectations: passed_records, failed_records; dropped_records is next to it
• By default the log is a hidden table, readable by the run-as user. Publish it to a named table in Advanced settings → event log
• Put a view over the published log before granting access to others

• Read the issues panel and the event log error first: it names the failing flow and the cause
• Refresh failed tables reruns the failed tables and what depends on them
• Select tables for refresh reruns a chosen subset. Excluded tables aren't refreshed, even if downstream
• Full refresh selection reprocesses data for chosen tables
• Validate (dry run) finds name and column errors without writing data

• Development runs (UI) reuse compute and don't retry, so errors show fast
• Job-triggered runs retry and restart for recoverable errors: use them for production (Module 10)