From 5e74d7d1fe8733e944d668a87c1d06ea3d1b25fe Mon Sep 17 00:00:00 2001 From: Anand Raman Date: Tue, 7 Jul 2026 17:29:22 -0500 Subject: [PATCH] Lakeflow Declarative Pipelines and Feldera Demo Added configuration for a Lakeflow declarative pipeline. Users can copy the bronze data to their own bucket, see how Feldera works with snapshot_and_follow mode, and contrast with Databricks IVM solution (lakeflow declarative pipelines) Signed-off-by: Anand Raman --- docs.feldera.com/docs/sidebars.js | 3 +- .../databricks/ingest_bronze.py | 311 +++++++++++ .../databricks/job.yaml | 37 ++ .../databricks/measure_latency.py | 87 +++ .../databricks/pipeline.yaml | 16 + .../databricks/silver_gold_pipeline.py | 497 ++++++++++++++++++ .../use_cases/medallion_architecture/part3.md | 2 + .../use_cases/medallion_architecture/part4.md | 305 +++++++++++ 8 files changed, 1257 insertions(+), 1 deletion(-) create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py create mode 100644 docs.feldera.com/docs/use_cases/medallion_architecture/part4.md diff --git a/docs.feldera.com/docs/sidebars.js b/docs.feldera.com/docs/sidebars.js index f7342618e17..fbbbe29773e 100644 --- a/docs.feldera.com/docs/sidebars.js +++ b/docs.feldera.com/docs/sidebars.js @@ -120,7 +120,8 @@ const guides = { items: [ 'use_cases/medallion_architecture/part1', 'use_cases/medallion_architecture/part2', - 'use_cases/medallion_architecture/part3' + 'use_cases/medallion_architecture/part3', + 'use_cases/medallion_architecture/part4' ] }, { diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py new file mode 100644 index 00000000000..e7156f89613 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/ingest_bronze.py @@ -0,0 +1,311 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # Step 1 — Ingest CDC into the Bronze Delta tables +# MAGIC +# MAGIC First task of the `compute_silver_gold` Databricks Workflow. It does two things: +# MAGIC +# MAGIC 1. **(optional) `clean_start`** — when the job parameter `clean_start = true`, the +# MAGIC working bronze Delta tables are deleted and re-seeded from the read-only source +# MAGIC snapshot at `s3://feldera-demos/ecommerce-cdc-{scale}/snapshot`. The silver and +# MAGIC gold materialized views are then recomputed by the pipeline refresh step — a +# MAGIC normal refresh, which Enzyme automatically turns into a full recompute once it +# MAGIC sees bronze was rewritten. +# MAGIC 2. **CDC ingest** — read the Feldera-format CDC NDJSON file for the single hour +# MAGIC `hour` and apply it to the working bronze Delta tables on S3. `bronze_orders` +# MAGIC carries status updates (delete-of-old + insert-of-new), so it is MERGEd by +# MAGIC `order_id` keeping the latest row; the other three fact tables are append-only +# MAGIC event streams, so their inserts are simply appended. +# MAGIC +# MAGIC Run the hours in order (`...T00`, `...T01`, ...): each run applies exactly one +# MAGIC hour of change, the same granularity at which `push_changes.py --hour` feeds +# MAGIC Feldera. +# MAGIC +# MAGIC After this notebook finishes, the workflow refreshes the silver/gold materialized +# MAGIC views defined in `silver_gold_pipeline.py`. +# MAGIC +# MAGIC **Feldera equivalent:** there is no separate ingest step — Feldera follows the +# MAGIC bronze Delta tables and propagates each change through the view DAG incrementally. + +# COMMAND ---------- + +from pyspark.sql import functions as F +from pyspark.sql.window import Window + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Parameters + +# COMMAND ---------- + +# CDC spans 7 days x 24 hours (generate_snapshot_cdc.py): 2025-11-30T00 .. 2025-12-06T23. +# Offer those as a picklist, plus "" for a seed-only run (no CDC). +from datetime import datetime, timedelta + +CDC_FIRST_DAY = datetime(2025, 11, 30) +CDC_NUM_DAYS = 7 +HOUR_CHOICES = [""] + [ + (CDC_FIRST_DAY + timedelta(hours=h)).strftime("%Y-%m-%dT%H") + for h in range(CDC_NUM_DAYS * 24) +] + +dbutils.widgets.dropdown("hour", "", HOUR_CHOICES, "hour (empty = seed only)") +dbutils.widgets.text("scale_factor", "0.01", "scale_factor") +dbutils.widgets.dropdown("clean_start", "false", ["false", "true"], "clean_start") +dbutils.widgets.text("source_bucket", "s3://feldera-demos", "source_bucket") +dbutils.widgets.text("warehouse_bucket", "s3://feldera-demos", "warehouse_bucket") + +HOUR = dbutils.widgets.get("hour").strip() or None +SCALE_FACTOR = float(dbutils.widgets.get("scale_factor")) +CLEAN_START = dbutils.widgets.get("clean_start").strip().lower() == "true" +SOURCE_BUCKET = dbutils.widgets.get("source_bucket").strip() +WAREHOUSE_BUCKET = dbutils.widgets.get("warehouse_bucket").strip() + +SCALE_STR = str(SCALE_FACTOR).replace(".", "-") + +# Read-only source: the snapshot + CDC produced by generate_snapshot_cdc.py. +# SOURCE_BUCKET comes from the source_bucket job parameter (read above). +SOURCE_PREFIX = f"ecommerce-cdc-{SCALE_STR}" +SNAPSHOT_ROOT = f"{SOURCE_BUCKET}/{SOURCE_PREFIX}/snapshot" +CDC_ROOT = f"{SOURCE_BUCKET}/{SOURCE_PREFIX}/cdc" + +# Writable working warehouse: the bronze tables this workflow ingests into and the +# silver/gold materialized-view pipeline reads from. Kept separate from the read-only +# source so a clean_start can rebuild it from scratch without touching the source. +# WAREHOUSE_BUCKET comes from the warehouse_bucket job parameter (read above), the same +# value the pipeline reads via spark.conf — single source of truth across both tasks. +WAREHOUSE_PREFIX = f"ecommerce-pipeline-{SCALE_STR}" +BRONZE_ROOT = f"{WAREHOUSE_BUCKET}/{WAREHOUSE_PREFIX}/bronze" + + +def s3a(path: str) -> str: + return path.replace("s3://", "s3a://") + + +print(f"clean_start: {CLEAN_START}") +print(f"hour: {HOUR or '(none — seed only)'}") +print(f"scale_factor: {SCALE_FACTOR}") +print(f"source: {SNAPSHOT_ROOT}") +print(f"cdc: {CDC_ROOT}") +print(f"bronze (rw): {BRONZE_ROOT}") + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Table metadata +# MAGIC +# MAGIC All seven bronze tables are re-seeded on `clean_start`. Only the four fact tables +# MAGIC receive CDC, and they split into two ingest modes: +# MAGIC +# MAGIC - **`merge`** — `bronze_orders` only. Orders go pending -> confirmed -> shipped -> +# MAGIC delivered; each transition is a delete-of-old + insert-of-new, so the hour's +# MAGIC inserts are deduped to the latest `updated_at` per `order_id` and MERGEd. +# MAGIC - **`append`** — order items, clickstream and inventory are append-only event +# MAGIC streams that never update an existing row, so the hour's inserts are appended +# MAGIC directly (no join against the target, no file rewrites). + +# COMMAND ---------- + +# All bronze tables. Used for the clean_start re-seed (DEEP CLONE from snapshot). +BRONZE_TABLES = [ + "bronze_suppliers", + "bronze_products", + "bronze_customers", + "bronze_orders", + "bronze_order_items", + "bronze_clickstream_events", + "bronze_inventory_events", +] + +# Fact tables that stream CDC -> how to apply one hour of change. +# merge: upsert keyed on `pk`, keeping the latest `order_by` row per key. +# append: insert-only; append the hour's new rows. +CDC_TABLES = { + "bronze_orders": { + "short": "orders", + "mode": "merge", + "pk": "order_id", + "order_by": "updated_at", + }, + "bronze_order_items": {"short": "order_items", "mode": "append"}, + "bronze_clickstream_events": {"short": "clickstream_events", "mode": "append"}, + "bronze_inventory_events": {"short": "inventory_events", "mode": "append"}, +} + + +def bronze_path(name: str) -> str: + return s3a(f"{BRONZE_ROOT}/{name}") + + +def snapshot_path(name: str) -> str: + return s3a(f"{SNAPSHOT_ROOT}/{name}") + + +def bronze_exists(name: str) -> bool: + """True if a Delta table is present at the working bronze path for `name` + (checks for its _delta_log). Uses the s3:// form for dbutils.fs.""" + try: + dbutils.fs.ls(f"{BRONZE_ROOT}/{name}/_delta_log") + return True + except Exception: # noqa: BLE001 — path absent => table not seeded + return False + + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## clean_start — delete and re-seed bronze from the source snapshot +# MAGIC +# MAGIC Deletes the working bronze path, then copies each bronze table fresh from the +# MAGIC read-only source snapshot via DEEP CLONE. Silver/gold are materialized views owned +# MAGIC by the Lakeflow pipeline (not stored at these warehouse paths), so they aren't +# MAGIC touched here — they recompute on the pipeline refresh step, which Enzyme turns into +# MAGIC a full recompute once it sees bronze was rewritten. + +# COMMAND ---------- + +if CLEAN_START: + print("clean_start = true -> deleting bronze re-seed") + layer_path = f"{WAREHOUSE_BUCKET}/{WAREHOUSE_PREFIX}/bronze" + try: + dbutils.fs.rm(layer_path, recurse=True) + print(f" deleted {layer_path}") + except Exception as e: # noqa: BLE001 — best-effort delete of a possibly-absent path + print(f" skip {layer_path}: {e}") + + for name in BRONZE_TABLES: + src = snapshot_path(name) + dst = bronze_path(name) + # DEEP CLONE copies the source data files directly (and their Delta stats) + # instead of reading every row back through Spark and re-encoding it on write. + # The row count comes from the operation metrics, so there is no extra scan. + # Enable row tracking on the working bronze: the silver/gold pipeline's incremental + # materialized views require row tracking on their source tables, else the refresh + # fails with ROW_TRACKING_NOT_ENABLED. Set on create so the later MERGE/append + # writes preserve it. + res = spark.sql( + f"CREATE OR REPLACE TABLE delta.`{dst}` " + f"DEEP CLONE delta.`{src}` " + "TBLPROPERTIES ('delta.enableRowTracking' = 'true')" + ).collect() + copied = res[0]["num_copied_files"] if res else 0 + print(f" re-seeded {name}: cloned {copied} data files") + print( + "clean_start complete — bronze tables are a fresh copy of the source snapshot" + ) +else: + print("clean_start = false -> keeping existing bronze tables, applying CDC on top") + # clean_start=false means we apply CDC on top of existing bronze, so both an hour to + # ingest and pre-seeded bronze tables are required. Fail fast with a clear message + # rather than no-op'ing here and erroring later in the pipeline's bronze reads. + if HOUR is None: + raise ValueError( + "hour is required when clean_start=false (nothing to ingest otherwise). " + "Pass hour=YYYY-MM-DDThh, or set clean_start=true to seed bronze first." + ) + missing = [t for t in BRONZE_TABLES if not bronze_exists(t)] + if missing: + raise FileNotFoundError( + f"Bronze tables missing under {BRONZE_ROOT}: {missing}. " + "Run once with clean_start=true to seed bronze from the source snapshot " + "before applying CDC." + ) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## CDC ingest — apply one hour of Feldera-format change records to bronze +# MAGIC +# MAGIC Each CDC line is `{"insert": {...}}` / `{"delete": {...}}` +# MAGIC ([Feldera JSON format](https://docs.feldera.com/formats/json)); a status update is +# MAGIC a delete of the old row followed by an insert of the new one. We keep the inserted +# MAGIC state (the new/updated row) and drop the deleted state — an update re-inserts the +# MAGIC new row, which wins via the orders MERGE key, so the inserts alone reconstruct the +# MAGIC current table. Only one hour's file (`{hour}.json`) is read per table. + +# COMMAND ---------- + + +def read_hour_inserts(table_short_name, target_schema, hour): + """Read the single CDC file for `hour` and return its inserted row state, cast to + the target (bronze) schema. Returns None if that hour has no file for this table.""" + path = s3a(f"{CDC_ROOT}/{table_short_name}/{hour}.json") + try: + # spark.read.json on a missing path throws; a present-but-empty file yields no + # rows. Either way we treat it as "no change this hour". + raw = spark.read.json(path) + except Exception: # noqa: BLE001 — no file for this hour/table + return None + if "insert" not in raw.columns: + return None + + records = raw.filter(F.col("insert").isNotNull()).select("insert.*") + if not records.head(1): + return None + + # JSON serializes decimals/timestamps as strings — cast back to the bronze schema. + for field in target_schema.fields: + if field.name in records.columns: + records = records.withColumn( + field.name, F.col(field.name).cast(field.dataType) + ) + + return records.select([f.name for f in target_schema.fields]) + + +def ingest_hour(table_name, hour): + """Apply one hour of CDC to `table_name`: MERGE for orders, append otherwise.""" + cfg = CDC_TABLES[table_name] + dst = bronze_path(table_name) + target_schema = spark.read.format("delta").load(dst).schema + + cdc = read_hour_inserts(cfg["short"], target_schema, hour) + if cdc is None: + print(f" {table_name}: no CDC for {hour} — unchanged") + return + + if cfg["mode"] == "append": + # Insert-only stream: append the hour's rows. No join against the target and no + # file rewrites — the cheapest possible Delta write. + cdc.write.format("delta").mode("append").save(dst) + print(f" {table_name}: appended {cdc.count():,} rows") + return + + # merge mode (orders): the hour may carry several transitions for one order_id; + # keep the latest updated_at, then upsert. MERGE returns a metrics row, so the + # counts come for free without a second pass over the source DAG. + pk, order_by = cfg["pk"], cfg["order_by"] + w = Window.partitionBy(pk).orderBy(F.col(order_by).desc()) + latest = ( + cdc.withColumn("_rn", F.row_number().over(w)) + .filter(F.col("_rn") == 1) + .drop("_rn") + ) + + latest.createOrReplaceTempView("cdc_src") + metrics = ( + spark.sql(f""" + MERGE INTO delta.`{dst}` AS t + USING cdc_src AS s + ON t.{pk} = s.{pk} + WHEN MATCHED THEN UPDATE SET * + WHEN NOT MATCHED THEN INSERT * + """) + .collect()[0] + .asDict() + ) + print( + f" {table_name}: {metrics.get('num_inserted_rows', 0):,} inserted, " + f"{metrics.get('num_updated_rows', 0):,} updated (keyed on {pk})" + ) + + +# COMMAND ---------- + +if HOUR is None: + print("hour is empty — skipping CDC ingest (bronze left at its seeded state).") +else: + print(f"Applying CDC hour {HOUR} to bronze...") + for table_name in CDC_TABLES: + ingest_hour(table_name, HOUR) + print("Bronze ingest complete. Silver/gold materialized views refresh next.") diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml new file mode 100644 index 00000000000..2d59db54473 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml @@ -0,0 +1,37 @@ +# Job settings for the compute_silver_gold workflow. +# Create via the Jobs UI (Create job -> kebab menu -> Edit as YAML) or the jobs API. +# - Replace the ingest_bronze notebook_path with where you imported ingest_bronze.py. +# - Replace pipeline_id with the id of the pipeline created from pipeline.yaml. +# - Assign serverless (or other) compute to the notebook task in the UI. +# scale_factor + warehouse_bucket are passed to the ingest notebook here and also override +# the pipeline's parameters of the same key when the pipeline task runs. +resources: + jobs: + name: compute_silver_gold + parameters: + - name: hour + default: "" + - name: clean_start + default: "false" + - name: scale_factor + default: "0.01" + - name: source_bucket + default: s3://feldera-demos + - name: warehouse_bucket + default: + tasks: + - task_key: ingest_bronze + notebook_task: + notebook_path: + base_parameters: + hour: "{{job.parameters.hour}}" + clean_start: "{{job.parameters.clean_start}}" + scale_factor: "{{job.parameters.scale_factor}}" + source_bucket: "{{job.parameters.source_bucket}}" + warehouse_bucket: "{{job.parameters.warehouse_bucket}}" + - task_key: refresh_silver_gold + depends_on: + - task_key: ingest_bronze + pipeline_task: + pipeline_id: "" + full_refresh: false diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py new file mode 100644 index 00000000000..c5b4a0bf37e --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/measure_latency.py @@ -0,0 +1,87 @@ +# Measures end-to-end latency for the Delta Lake input connectors using the +# completed_frontier timestamps exposed on /stats. +# See https://docs.feldera.com/pipelines/latency/#measuring-end-to-end-latency-with-input-frontiers +# +# For each Delta connector the frontier reports three RFC 3339 timestamps: +# ingested_at -- data for this frontier arrived at the connector +# processed_at -- the IVM engine finished processing it +# completed_at -- outputs reached all sinks +# Latency is the delta between these, plus freshness = now - completed_at. +import os +from argparse import ArgumentParser +from datetime import datetime, timezone + +from dotenv import load_dotenv +from feldera import FelderaClient + +PIPELINE = "ecommerce-medallion-architecture" + +# Bronze tables to report on (user-facing names map to bronze_). +TABLES = [ + "bronze_clickstream_events", + "bronze_inventory_events", + "bronze_orders", + "bronze_order_items", +] + + +def parse_ts(value: str) -> datetime: + # Frontier timestamps are RFC 3339 with a trailing Z; normalize for fromisoformat. + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def delta_connector(inputs: list[dict], table: str) -> dict | None: + # A Delta table has both an api-ingress and a delta_table_input connector on + # the same stream; only the Delta one populates completed_frontier. + for i in inputs: + if i["endpoint_name"].startswith(f"{table}.") and i.get("completed_frontier"): + return i + return None + + +def report(client: FelderaClient) -> None: + stats = client.get_pipeline_stats(PIPELINE) + now = datetime.now(timezone.utc) + + header = ( + f"{'table':<26}{'version':>8}{'ingest→proc':>14}{'proc→done':>12}{'e2e':>10}" + ) + print(header) + print("-" * len(header)) + + for table in TABLES: + conn = delta_connector(stats["inputs"], table) + if conn is None: + print(f"{table:<26}{'no delta frontier yet':>56}") + continue + + f = conn["completed_frontier"] + version = f.get("metadata", {}).get("version") + ingested = parse_ts(f["ingested_at"]) + processed = parse_ts(f["processed_at"]) + completed = parse_ts(f["completed_at"]) + + ingest_to_proc = (processed - ingested).total_seconds() + proc_to_done = (completed - processed).total_seconds() + e2e = (completed - ingested).total_seconds() + + print( + f"{table:<26}{version:>8}" + f"{ingest_to_proc:>13.2f}s{proc_to_done:>11.2f}s" + f"{e2e:>9.2f}s" + ) + + +if __name__ == "__main__": + parser = ArgumentParser( + description="Measure Delta input latency via /stats frontiers" + ) + parser.add_argument("--pipeline", default=PIPELINE) + args = parser.parse_args() + PIPELINE = args.pipeline + + load_dotenv() + client = FelderaClient( + url=os.getenv("FELDERA_URL"), api_key=os.getenv("FELDERA_API_KEY") + ) + report(client) diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml new file mode 100644 index 00000000000..4b8e08f9c50 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml @@ -0,0 +1,16 @@ +# Lakeflow Declarative Pipeline settings (the "transformation" — runs silver_gold_pipeline.py). +# Create via the Pipelines UI (Create pipeline -> Settings -> YAML) or the pipelines API. +# - Replace the notebook path with where you imported silver_gold_pipeline.py. +# - Set catalog/schema to where the silver/gold materialized views should publish. +# configuration values are read via spark.conf in the notebook; the compute_silver_gold job +# passes job parameters of the same key, which override these at run time. +name: ecommerce-silver-gold-mv +serverless: true +catalog: +schema: +configuration: + scale_factor: "0.01" + warehouse_bucket: +libraries: + - notebook: + path: diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py new file mode 100644 index 00000000000..632e1accaf0 --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/silver_gold_pipeline.py @@ -0,0 +1,497 @@ +# Databricks notebook source +# MAGIC %md +# MAGIC # Silver & Gold — Databricks Materialized Views (Lakeflow Declarative Pipeline) +# MAGIC +# MAGIC The silver and gold layers from `compute_silver_gold.py`, expressed as a Lakeflow +# MAGIC Declarative Pipeline. Each silver/gold dataset is a **materialized view** that +# MAGIC Databricks refreshes incrementally where the query allows (Enzyme), falling back +# MAGIC to a full recompute otherwise. +# MAGIC +# MAGIC - **Bronze** is exposed as lightweight `@dp.temporary_view`s that read the working +# MAGIC bronze Delta tables on S3 (written by `ingest_bronze.py`). They are not persisted — +# MAGIC the bronze tables already live in S3. +# MAGIC - **Silver / Gold** are `@dp.materialized_view`s, persisted in the pipeline's Unity +# MAGIC Catalog target schema. +# MAGIC +# MAGIC The business logic (SQL) is identical to the batch script and to the Feldera views +# MAGIC in `ecommerce/sql/ecommerce.sql`. The workflow refreshes this pipeline as its second +# MAGIC step with a single normal refresh: it updates incrementally after a CDC hour, and +# MAGIC Enzyme automatically full-recomputes after a `clean_start` rewrites bronze. +# MAGIC +# MAGIC **Feldera equivalent:** the same DAG, maintained incrementally on every commit with +# MAGIC no pipeline scheduling and no full-refresh fallback. + +# COMMAND ---------- + +from pyspark import pipelines as dp +from pyspark.sql import functions as F + +# scale_factor is supplied as pipeline configuration; it selects the bronze S3 path the +# ingest step wrote to. Defaults to 0.01 to match the rest of the demo. +SCALE_FACTOR = float(spark.conf.get("scale_factor", "0.01")) +SCALE_STR = str(SCALE_FACTOR).replace(".", "-") + +WAREHOUSE_BUCKET = spark.conf.get("warehouse_bucket") +BRONZE_ROOT = f"{WAREHOUSE_BUCKET}/ecommerce-pipeline-{SCALE_STR}/bronze".replace( + "s3://", "s3a://" +) + +# The silver/gold materialized views publish to the pipeline's configured catalog/schema +# (set on the pipeline when it's created — see pipeline.yaml catalog/schema). Datasets use +# bare names and resolve within that destination. + + +def bronze_path(name: str) -> str: + return f"{BRONZE_ROOT}/{name}" + + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Bronze source views +# MAGIC +# MAGIC Pass-through views over the working bronze Delta tables on S3. Declaring them as +# MAGIC pipeline datasets lets the silver/gold materialized views reference them by name +# MAGIC (in `spark.sql`) and lets Enzyme reason about their changes. + +# COMMAND ---------- + +_BRONZE = [ + "bronze_suppliers", + "bronze_products", + "bronze_customers", + "bronze_orders", + "bronze_order_items", + "bronze_clickstream_events", + "bronze_inventory_events", +] + + +def _make_bronze_view(table_name): + @dp.temporary_view(name=table_name) + def _view(): + return spark.read.format("delta").load(bronze_path(table_name)) + + return _view + + +for _name in _BRONZE: + _make_bronze_view(_name) + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Silver layer +# MAGIC +# MAGIC Each function runs the exact SQL from `compute_silver_gold.py`, referencing its +# MAGIC upstream pipeline datasets directly by name — Lakeflow resolves those names against +# MAGIC the pipeline graph and registers them as dependencies, so the logic stays in +# MAGIC lock-step with the batch script and the Feldera views. + +# COMMAND ---------- + + +@dp.materialized_view( + name="silver_customers", + comment="Cleaned customer dimension (valid tiers).", + refresh_policy="incremental", +) +def silver_customers(): + return spark.read.table("bronze_customers").filter( + F.col("customer_id").isNotNull() + & F.col("customer_tier").isin("standard", "silver", "gold", "platinum") + ) + + +@dp.materialized_view( + name="silver_products", + comment="Cleaned product dimension (product grain).", + refresh_policy="incremental", +) +def silver_products(): + return spark.sql(""" + SELECT product_id, product_name, category, brand, list_price + FROM bronze_products + WHERE product_id IS NOT NULL + """) + + +@dp.materialized_view( + name="silver_enriched_clickstream", + comment="Clickstream filtered to interaction events (clean-only).", + refresh_policy="incremental", +) +def silver_enriched_clickstream(): + return spark.sql(""" + SELECT + ce.event_id, ce.user_id, ce.session_id, ce.event_type, ce.page_url, + ce.product_id, ce.device_type, ce.geo_country, ce.geo_region, + ce.event_timestamp + FROM bronze_clickstream_events ce + WHERE ce.event_type IS NOT NULL + AND ce.user_id IS NOT NULL + AND ce.event_type IN ('page_view', 'product_view', 'add_to_cart', 'begin_checkout', 'purchase') + """) + + +@dp.materialized_view( + name="silver_orders_enriched", + comment="Orders joined to customers + aggregated order items.", + refresh_policy="incremental", +) +def silver_orders_enriched(): + return spark.sql(""" + SELECT + o.order_id, o.user_id, o.order_status, o.order_total, + o.discount_amount, o.shipping_cost, o.payment_method, o.coupon_code, + o.created_at, o.updated_at, + c.customer_tier, + c.geo_country AS customer_country, + c.geo_region AS customer_region, + c.signup_date, + oi.item_count, oi.total_quantity, oi.gross_item_revenue, oi.avg_discount_pct + FROM bronze_orders o + JOIN silver_customers c ON o.user_id = c.customer_id + JOIN ( + SELECT order_id, + COUNT(*) AS item_count, + SUM(quantity) AS total_quantity, + SUM(quantity * unit_price) AS gross_item_revenue, + AVG(discount_pct) AS avg_discount_pct + FROM bronze_order_items + WHERE quantity > 0 AND unit_price >= 0 + GROUP BY order_id + ) oi ON o.order_id = oi.order_id + WHERE o.order_id IS NOT NULL + AND o.user_id IS NOT NULL + AND o.order_total >= 0 + AND o.order_status IN ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'returned') + """) + + +@dp.materialized_view( + name="silver_order_items_enriched", + comment="Order items joined to products, suppliers, orders, customers with margins.", + refresh_policy="incremental", +) +def silver_order_items_enriched(): + return spark.sql(""" + SELECT + oi.order_item_id, oi.order_id, oi.product_id, + oi.quantity, oi.unit_price, oi.discount_pct, + oi.quantity * oi.unit_price AS line_gross_revenue, + oi.quantity * oi.unit_price * (1.0 - COALESCE(oi.discount_pct, 0) / 100.0) AS line_net_revenue, + oi.quantity * (oi.unit_price - p.cost_price) AS line_gross_margin, + p.product_name, p.category, p.brand, p.cost_price, p.list_price, + s.supplier_id, s.supplier_name, s.country AS supplier_country, s.lead_time_days, + o.created_at AS order_created_at, + o.order_status, o.user_id, + c.customer_tier + FROM bronze_order_items oi + JOIN bronze_products p ON oi.product_id = p.product_id + JOIN bronze_suppliers s ON p.supplier_id = s.supplier_id + JOIN bronze_orders o ON oi.order_id = o.order_id + JOIN silver_customers c ON o.user_id = c.customer_id + WHERE oi.quantity > 0 + AND oi.unit_price >= 0 + AND p.is_active = TRUE + AND o.order_status IN ('pending', 'confirmed', 'shipped', 'delivered', 'cancelled', 'returned') + """) + + +@dp.materialized_view( + name="silver_confirmed_order_items", + comment="Order items excluding cancelled/returned orders.", + refresh_policy="incremental", +) +def silver_confirmed_order_items(): + return spark.sql(""" + SELECT * FROM silver_order_items_enriched + WHERE order_status NOT IN ('cancelled', 'returned') + """) + + +@dp.materialized_view( + name="silver_inventory_current", + comment="Running stock per product/warehouse from inventory events.", + refresh_policy="incremental", +) +def silver_inventory_current(): + return spark.sql(""" + SELECT + ie.product_id, ie.warehouse_id, + p.product_name, p.category, p.brand, + s.supplier_id, s.supplier_name, s.lead_time_days, + SUM(ie.quantity_change) AS current_stock, + SUM(CASE WHEN ie.event_type = 'restock' THEN ie.quantity_change ELSE 0 END) AS total_restocked, + SUM(CASE WHEN ie.event_type = 'sale_reserve' THEN ABS(ie.quantity_change) ELSE 0 END) + - SUM(CASE WHEN ie.event_type = 'cancellation_restock' THEN ie.quantity_change ELSE 0 END) + AS total_sold, + SUM(CASE WHEN ie.event_type = 'return_restock' THEN ie.quantity_change ELSE 0 END) AS total_returned + FROM bronze_inventory_events ie + JOIN bronze_products p ON ie.product_id = p.product_id + JOIN bronze_suppliers s ON p.supplier_id = s.supplier_id + WHERE ie.quantity_change IS NOT NULL AND ie.product_id IS NOT NULL + GROUP BY ie.product_id, ie.warehouse_id, p.product_name, p.category, p.brand, s.supplier_id, s.supplier_name, s.lead_time_days + """) + + +@dp.materialized_view( + name="silver_inventory_by_supplier", + comment="Stock rolled up to supplier grain.", + refresh_policy="incremental", +) +def silver_inventory_by_supplier(): + return spark.sql(""" + SELECT supplier_id, supplier_name, + SUM(current_stock) AS total_current_stock, + SUM(total_sold) AS total_sold, + SUM(total_restocked) AS total_restocked, + SUM(total_returned) AS total_returned + FROM silver_inventory_current + GROUP BY supplier_id, supplier_name + """) + + +# COMMAND ---------- + +# MAGIC %md +# MAGIC ## Gold layer + +# COMMAND ---------- + + +@dp.materialized_view( + name="gold_order_status_summary", + comment="Order status distribution.", + refresh_policy="incremental", +) +def gold_order_status_summary(): + return spark.sql(""" + SELECT + o.order_status, + COUNT(DISTINCT o.order_id) AS order_count, + COALESCE(SUM(o.order_total), 0) AS total_revenue, + AVG(o.order_total) AS avg_order_value, + COUNT(DISTINCT o.user_id) AS unique_customers + FROM silver_orders_enriched o + GROUP BY o.order_status + """) + + +@dp.materialized_view( + name="gold_supplier_performance", + comment="Revenue + margin per supplier.", + refresh_policy="incremental", +) +def gold_supplier_performance(): + return spark.sql(""" + WITH orders_by_supplier AS ( + SELECT + oi.supplier_id, oi.supplier_name, oi.supplier_country, oi.lead_time_days, + SUM(oi.quantity) AS total_units_sold, + SUM(oi.line_net_revenue) AS total_net_revenue, + SUM(oi.line_gross_margin) AS total_gross_margin, + SUM(oi.line_gross_margin) / NULLIF(SUM(oi.line_net_revenue), 0) AS avg_margin_pct, + AVG(oi.discount_pct) AS avg_discount_applied, + CAST(SUM(CASE WHEN oi.order_status = 'delivered' THEN 1 ELSE 0 END) AS DOUBLE) + / NULLIF(CAST(COUNT(*) AS DOUBLE), 0) AS reliability_score + FROM silver_confirmed_order_items oi + GROUP BY oi.supplier_id, oi.supplier_name, oi.supplier_country, oi.lead_time_days + ), + products_by_supplier AS ( + SELECT supplier_id, COUNT(*) AS products_sold + FROM (SELECT DISTINCT supplier_id, product_id FROM silver_confirmed_order_items) + GROUP BY supplier_id + ), + orders_count_by_supplier AS ( + SELECT supplier_id, COUNT(*) AS orders_fulfilled + FROM (SELECT DISTINCT supplier_id, order_id FROM silver_confirmed_order_items) + GROUP BY supplier_id + ) + SELECT + oi.supplier_id, oi.supplier_name, oi.supplier_country, oi.lead_time_days, + p.products_sold, + o.orders_fulfilled, + oi.total_units_sold, + oi.total_net_revenue, + oi.total_gross_margin, + oi.avg_margin_pct, + oi.avg_discount_applied, + oi.reliability_score, + inv.total_current_stock, + inv.total_sold AS inventory_units_sold, + inv.total_restocked AS inventory_units_restocked, + inv.total_returned AS inventory_units_returned + FROM orders_by_supplier oi + JOIN silver_inventory_by_supplier inv ON oi.supplier_id = inv.supplier_id + JOIN products_by_supplier p ON oi.supplier_id = p.supplier_id + JOIN orders_count_by_supplier o ON oi.supplier_id = o.supplier_id + """) + + +@dp.materialized_view( + name="gold_inventory_risk", + comment="Stock risk scoring per product.", + refresh_policy="incremental", +) +def gold_inventory_risk(): + return spark.sql(""" + SELECT + inv.product_id, inv.product_name, inv.category, inv.brand, + inv.supplier_name, inv.lead_time_days, + inv.total_stock_all_warehouses, + sales.units_sold, sales.avg_daily_units, + CASE WHEN sales.avg_daily_units > 0 + THEN inv.total_stock_all_warehouses / sales.avg_daily_units ELSE NULL END AS days_of_stock_remaining, + CASE + WHEN sales.avg_daily_units > 0 AND (inv.total_stock_all_warehouses / sales.avg_daily_units) < inv.lead_time_days * 1.5 THEN 'CRITICAL' + WHEN sales.avg_daily_units > 0 AND (inv.total_stock_all_warehouses / sales.avg_daily_units) < inv.lead_time_days * 3.0 THEN 'WARNING' + ELSE 'OK' + END AS stock_risk_level, + sales.net_revenue, sales.gross_margin + FROM ( + SELECT product_id, product_name, category, brand, supplier_name, lead_time_days, + SUM(current_stock) AS total_stock_all_warehouses + FROM silver_inventory_current + GROUP BY product_id, product_name, category, brand, supplier_name, lead_time_days + ) inv + JOIN ( + SELECT product_id, + SUM(quantity) AS units_sold, + SUM(quantity) / NULLIF(COUNT(DISTINCT date_trunc('day', order_created_at)), 0) AS avg_daily_units, + SUM(line_net_revenue) AS net_revenue, + SUM(line_gross_margin) AS gross_margin + FROM silver_confirmed_order_items + GROUP BY product_id + ) sales ON inv.product_id = sales.product_id + """) + + +@dp.materialized_view( + name="gold_realtime_inventory_alerts", + comment="CRITICAL stock filter.", + refresh_policy="incremental", +) +def gold_realtime_inventory_alerts(): + return spark.sql( + "SELECT * FROM gold_inventory_risk WHERE stock_risk_level = 'CRITICAL'" + ) + + +@dp.materialized_view( + name="gold_weekly_revenue_trend", + comment="Weekly revenue with WoW change, moving avg, YTD (window functions).", + refresh_policy="incremental", +) +def gold_weekly_revenue_trend(): + return spark.sql(""" + SELECT + week_start, category, + weekly_net_revenue, weekly_gross_margin, order_count, units_sold, + weekly_net_revenue - LAG(weekly_net_revenue, 1) OVER (PARTITION BY category ORDER BY week_start) + AS revenue_wow_change, + (weekly_net_revenue - LAG(weekly_net_revenue, 1) OVER (PARTITION BY category ORDER BY week_start)) + / NULLIF(LAG(weekly_net_revenue, 1) OVER (PARTITION BY category ORDER BY week_start), 0) + AS revenue_wow_pct_change, + AVG(weekly_net_revenue) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) + AS revenue_4wk_moving_avg, + AVG(weekly_gross_margin) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) + AS margin_4wk_moving_avg, + SUM(weekly_net_revenue) OVER (PARTITION BY category, EXTRACT(YEAR FROM week_start) ORDER BY week_start RANGE UNBOUNDED PRECEDING) + AS cumulative_ytd_revenue + FROM ( + SELECT + date_trunc('week', oi.order_created_at) AS week_start, + oi.category, + SUM(oi.line_net_revenue) AS weekly_net_revenue, + SUM(oi.line_gross_margin) AS weekly_gross_margin, + COUNT(DISTINCT oi.order_id) AS order_count, + SUM(oi.quantity) AS units_sold + FROM silver_confirmed_order_items oi + GROUP BY date_trunc('week', oi.order_created_at), oi.category + ) + """) + + +@dp.materialized_view( + name="gold_cancellation_impact", + comment="Cancellation rate windows (running totals).", + refresh_policy="incremental", +) +def gold_cancellation_impact(): + return spark.sql(""" + SELECT + category, week_start, + weekly_cancelled_orders, weekly_total_orders, + weekly_cancelled_revenue, weekly_total_revenue, + CAST(SUM(weekly_cancelled_orders) OVER (PARTITION BY category ORDER BY week_start RANGE UNBOUNDED PRECEDING) AS DOUBLE) + / NULLIF(CAST(SUM(weekly_total_orders) OVER (PARTITION BY category ORDER BY week_start RANGE UNBOUNDED PRECEDING) AS DOUBLE), 0) + AS cumulative_cancellation_rate, + SUM(weekly_cancelled_revenue) OVER (PARTITION BY category ORDER BY week_start RANGE UNBOUNDED PRECEDING) + AS cumulative_cancelled_revenue, + CAST(SUM(weekly_cancelled_orders) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) AS DOUBLE) + / NULLIF(CAST(SUM(weekly_total_orders) OVER (PARTITION BY category ORDER BY week_start RANGE BETWEEN INTERVAL 3 WEEKS PRECEDING AND CURRENT ROW) AS DOUBLE), 0) + AS cancellation_rate_4wk + FROM ( + SELECT + oi.category, + date_trunc('week', oi.order_created_at) AS week_start, + COUNT(DISTINCT CASE WHEN oi.order_status = 'cancelled' THEN oi.order_id END) AS weekly_cancelled_orders, + COUNT(DISTINCT oi.order_id) AS weekly_total_orders, + SUM(CASE WHEN oi.order_status = 'cancelled' THEN oi.line_net_revenue ELSE 0 END) AS weekly_cancelled_revenue, + SUM(oi.line_net_revenue) AS weekly_total_revenue + FROM silver_order_items_enriched oi + GROUP BY oi.category, date_trunc('week', oi.order_created_at) + ) + """) + + +@dp.materialized_view( + name="gold_product_demand_surge", + comment="Trailing-24h cart velocity vs stock & lead time (sliding window).", + refresh_policy="auto", +) +def gold_product_demand_surge(): + return spark.sql(""" + WITH recent_demand AS ( + SELECT + ce.product_id, + SUM(CASE WHEN ce.event_type = 'product_view' THEN 1 ELSE 0 END) AS recent_views, + SUM(CASE WHEN ce.event_type = 'add_to_cart' THEN 1 ELSE 0 END) AS recent_add_to_carts, + SUM(CASE WHEN ce.event_type = 'begin_checkout' THEN 1 ELSE 0 END) AS recent_checkouts + FROM silver_enriched_clickstream ce + WHERE ce.product_id IS NOT NULL + AND ce.event_timestamp > + (SELECT MAX(event_timestamp) FROM silver_enriched_clickstream) - INTERVAL 1 DAY + GROUP BY ce.product_id + ), + stock_by_product AS ( + SELECT product_id, + SUM(current_stock) AS total_stock_all_warehouses, + MAX(lead_time_days) AS lead_time_days + FROM silver_inventory_current + GROUP BY product_id + ) + SELECT + d.product_id, p.product_name, p.category AS product_category, p.brand AS product_brand, + d.recent_views, d.recent_add_to_carts, d.recent_checkouts, + COALESCE(s.total_stock_all_warehouses, 0) AS total_stock_all_warehouses, + s.lead_time_days, + CAST(GREATEST(COALESCE(s.total_stock_all_warehouses, 0), 0) AS DOUBLE) + / NULLIF(CAST(d.recent_add_to_carts AS DOUBLE), 0) AS days_of_stock_cover, + CAST(d.recent_checkouts AS DOUBLE) + / NULLIF(CAST(d.recent_views AS DOUBLE), 0) AS view_to_checkout_rate, + CASE + WHEN d.recent_add_to_carts = 0 THEN 'OK' + WHEN COALESCE(s.total_stock_all_warehouses, 0) <= 0 THEN 'SURGE_STOCKOUT_RISK' + WHEN CAST(GREATEST(s.total_stock_all_warehouses, 0) AS DOUBLE) / CAST(d.recent_add_to_carts AS DOUBLE) + < COALESCE(s.lead_time_days, 7) THEN 'SURGE_STOCKOUT_RISK' + WHEN CAST(GREATEST(s.total_stock_all_warehouses, 0) AS DOUBLE) / CAST(d.recent_add_to_carts AS DOUBLE) + < COALESCE(s.lead_time_days, 7) * 2 THEN 'WARNING' + ELSE 'OK' + END AS demand_alert + FROM recent_demand d + LEFT JOIN silver_products p ON d.product_id = p.product_id + LEFT JOIN stock_by_product s ON d.product_id = s.product_id + """) diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md b/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md index 030ace38d67..e304f3f6e27 100644 --- a/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/part3.md @@ -51,3 +51,5 @@ Some of the silver views could be made incremental by hand by a data engineer More complex views, such as `gold_weekly_revenue_trend`, would require merge logic that is difficult to validate and entails significant engineering effort. With Feldera, you use the same SQL your batch engine runs now, get the same results, and pay a cost that tracks change size instead of data size. In addition, you get millisecond update latency across your full medallion architecture. + +Continue on to [Part 4](./part4.md) for a comparison between Databricks' IVM offering (Lakeflow Declarative Pipelines) and Feldera. \ No newline at end of file diff --git a/docs.feldera.com/docs/use_cases/medallion_architecture/part4.md b/docs.feldera.com/docs/use_cases/medallion_architecture/part4.md new file mode 100644 index 00000000000..f34a3266b5d --- /dev/null +++ b/docs.feldera.com/docs/use_cases/medallion_architecture/part4.md @@ -0,0 +1,305 @@ +# Part 4: Compare Feldera and Databricks Lakeflow Declarative Pipelines (Enzyme) + +Part [1](./part1.md) deploys the Feldera pipeline, which reads from a public bucket. [Part 2](./part2.md) demonstrates change data processing by pushing data to the Feldera pipeline via HTTP. [Part 3](./part3.md) compares Feldera and Spark, showcasing the difference between incremental view maintenance and batch refreshes. Part 4 compares Feldera with Databricks Enzyme engine. In this section, you will deploy the Databricks assets into **your own** workspace +and data in **your own** S3 bucket, then repoint the Feldera pipeline at your bucket bucket. Two things +run against the same Bronze Delta tables: + +- A **Lakeflow Declarative Pipeline** powered by Databricks Enzyme engine, maintains the Silver and Gold layers as Databricks + materialized views. Lakeflow Declarative Pipelines are the Databricks implementation of Incremental View Maintenance. The Declarative Pipeline uses the same SQL as the Feldera Pipeline. To fairly compare what Databricks **can** incrementalize, the [`refresh_policy`](https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-materialized-view-refresh-policy) for views in the pipeline is set to `incremental` where possible. Databricks currently cannot incrementalize the `gold_product_demand_surge` view, so the policy for this view is `auto`. +- An **`ingest_bronze` job** acts as a **data-generating process**: each run writes one + hour of CDC into the Bronze Delta tables. Point Feldera at those same tables and it + follows the Delta log, updating Gold sub-second after each write. The + Databricks pipeline runs immediately after ingestion. + +`ingest_bronze` stands in for a real operational source continually landing data. Instead +of `push_changes.py` pushing JSON to Feldera's HTTP ingress (Part 2), the writer here is a +Databricks job writing to Delta on S3, and Feldera consumes those Delta commits directly. + +## Files + +All four assets live in +[`medallion_architecture/databricks/`](https://github.com/feldera/feldera/tree/main/docs.feldera.com/docs/use_cases/medallion_architecture/databricks). + +| File | Role | +|---|---| +| `silver_gold_pipeline.py` | Pipeline source notebook — 8 Silver + 7 Gold materialized views (Bronze exposed as views). | +| `pipeline.yaml` | Settings for the Lakeflow pipeline that runs `silver_gold_pipeline.py` (Step 1). | +| `ingest_bronze.py` | Job notebook — optional `clean_start` re-seed, then apply one hour of CDC to the Bronze Delta tables. | +| `job.yaml` | Settings for the `compute_silver_gold` job (Step 2). | +| `measure_latency.py` | Reports Feldera's per-connector processing and end-to-end latency for changes (Step 4). | + +## Prerequisites +To run this demo yourself, you will need: + +- A Databricks workspace with **Unity Catalog** and **serverless** enabled. +- Read access to the read-only source bucket `s3://feldera-demos/ecommerce-cdc-{scale}` + (the snapshot and CDC files, anonymous read in `us-west-1`). +- **A writable S3 bucket you provision** for the working Bronze tables. The Databricks + job writes here and the Feldera pipeline reads from here, so both Databricks and Feldera + need access to it. This bucket is yours — you cannot write to `s3://feldera-demos`. + +## Storage layout + +| Layer | Location | Written by | +|---|---|---| +| Source snapshot + CDC (read-only) | `s3://feldera-demos/ecommerce-cdc-{scale}/{snapshot,cdc}` || +| Working Bronze (read/write) | `s3:///ecommerce-pipeline-{scale}/bronze` | `ingest_bronze.py` | +| Silver / Gold materialized views | Unity Catalog `{catalog}.{schema}` | the Lakeflow pipeline, `silver_gold_pipeline.py`| + +`{scale}` is the `scale_factor` with the dot replaced by a dash, e.g. `0.01` → `0-01`. The +working Bronze is kept separate from the read-only source so a `clean_start` can rebuild it +from scratch without touching the source snapshot. + +## Step 1 — Create the Lakeflow pipeline + +Create the pipeline first: the job's second task references it by ID. + +1. **Import the pipeline source.** Workspace → your folder → **Import** → add + `silver_gold_pipeline.py`. It imports as a notebook (the `# Databricks notebook source` + header makes the `# COMMAND ----------` cells render). +2. **Create the pipeline.** Sidebar → **Jobs & Pipelines** → **Create → ETL pipeline** + (Lakeflow Declarative Pipeline): + - **Source code**: the imported `silver_gold_pipeline.py`. + - **Serverless**: on. + - **Destination**: Unity Catalog → set **catalog** and **schema** where the Silver/Gold + materialized views publish (e.g. `main` / `ecommerce_demo`). + - **Advanced → Configuration**: add the keys the notebook reads via `spark.conf`: + - `scale_factor` = `0.01` + - `warehouse_bucket` = `s3://` + - The full settings are in [**`pipeline.yaml`**](https://github.com/feldera/feldera/tree/main/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/pipeline.yaml) (paste into *Settings → YAML*, or + `POST /api/2.0/pipelines`, instead of filling the form by hand). +3. **Save and copy the pipeline ID** (Pipeline details, or the URL) — Step 2 needs it. You + don't have to run it standalone; the job triggers it. + +## Step 2 — Upload `ingest_bronze.py` and create the job + +1. **Import the ingest notebook.** Workspace → **Import** → add `ingest_bronze.py`. +2. **Create the job.** Sidebar → **Jobs & Pipelines** → **Create job**, name it + `compute_silver_gold`. Full settings are in [**`job.yaml`**](https://github.com/feldera/feldera/tree/main/docs.feldera.com/docs/use_cases/medallion_architecture/databricks/job.yaml) (or use *Edit as YAML* / the + jobs API). Configure: + - **Job parameters**: `hour` = `""`, `clean_start` = `false`, `scale_factor` = `0.01`, + `source_bucket` = `s3://feldera-demos`, `warehouse_bucket` = `s3://`. + - **Task 1 — `ingest_bronze`** (Notebook): select `ingest_bronze.py`, compute + **Serverless**. Under the task **Parameters**, map each to the job parameter: + `hour` = `{{job.parameters.hour}}`, `clean_start` = `{{job.parameters.clean_start}}`, + `scale_factor` = `{{job.parameters.scale_factor}}`, + `source_bucket` = `{{job.parameters.source_bucket}}`, + `warehouse_bucket` = `{{job.parameters.warehouse_bucket}}`. + - **Task 2 — `refresh_silver_gold`** (Pipeline): **Depends on** `ingest_bronze`; select + the pipeline from Step 1 (paste its **pipeline ID** into `job.yaml`'s + `pipeline_task.pipeline_id`). Full refresh off. + +`scale_factor` and `warehouse_bucket` appear in both the pipeline configuration (Step 1) +and the job parameters (Step 2). The job passes them to the ingest notebook and, because +the keys match the pipeline's configuration, overrides the pipeline's values at run time — +so keep the two in sync when you change them. + +## Step 3 — `ingest_bronze` as a data-generating process + +Each run of the `ingest_bronze` task writes one hour of change into the Bronze Delta tables +under `s3:///ecommerce-pipeline-{scale}/bronze`: + +- `bronze_orders` carries status updates (delete-of-old + insert-of-new), so the hour's + rows are deduped to the latest `updated_at` per `order_id` and **merged**. +- `bronze_order_items`, `bronze_clickstream_events`, and `bronze_inventory_events` are + append-only event streams, so the hour's rows are **appended**. + +Every run commits new versions to the Delta log. That log is exactly what Feldera follows. + +**Run now → Run with parameters:** + +- **First run** — seed Bronze from the source snapshot into your bucket, then the + materialized views compute on the refresh step: + `clean_start = true`. +- **Each later run** — write ONE hour of CDC, in order: + `hour = 2025-11-30T00` (`clean_start = false`), then `2025-11-30T01`, and so on. + +The hours available are 7 days × 24, starting with `2025-11-30T00`. Run hours in order, once each — see [Notes](#notes). + +## Step 4 — Point the Feldera pipeline at your bucket + +The Feldera pipeline from Part 1 reads the Bronze snapshot from the public demo bucket in +`snapshot` mode. Repoint it at the Bronze tables that `ingest_bronze` writes and switch to +`snapshot_and_follow`, so Feldera loads the seeded snapshot and then follows the Delta log +as each `ingest_bronze` run commits a new hour. See the +[Delta input connector](/connectors/sources/delta) reference for the mode options. + +Edit each Bronze table's connector in the pipeline SQL. For the four fact tables that +receive CDC, change the `uri` to your bucket and set `mode` to `snapshot_and_follow`: + +```json +{ + "transport": { + "name": "delta_table_input", + "config": { + "uri": "s3:///ecommerce-pipeline-0-01/bronze/bronze_orders", + "mode": "snapshot_and_follow", + "aws_region": "", + "transaction_mode": "catchup" + } + } +} +``` + +| Bronze table | Mode | Why | +|---|---|---| +| `bronze_orders`, `bronze_order_items`, `bronze_clickstream_events`, `bronze_inventory_events` | `snapshot_and_follow` | Receive CDC every `ingest_bronze` run — follow the Delta log. | +| `bronze_suppliers`, `bronze_products`, `bronze_customers` | `snapshot` | Dimensions, seeded once and not updated by CDC. | + +Point every table's `uri` at `s3:///ecommerce-pipeline-0-01/bronze/` +(the `0-01` segment is `scale_factor` `0.01` with the dot replaced by a dash). + +### Configure AWS authentication for a private bucket + +The demo bucket is public, so Parts 1–3 use `"aws_skip_signature": "true"`. If your +bucket is private, **drop `aws_skip_signature`** and supply credentials instead. The +simplest option is an access key on the connector config (`aws_region` is required — the +Delta library does not auto-detect it): + +```json +"aws_access_key_id": "", +"aws_secret_access_key": "", +"aws_region": "" +``` + +For all supported options — access keys, session tokens, instance/container roles, KMS +encryption, custom endpoints — see +[configuring AWS authentication for the Delta connector](/connectors/sources/delta#storage-parameters) +and the [Setting AWS credentials example](/connectors/sources/delta#example-setting-aws-credentials). + +### Run order + +1. Run `ingest_bronze` once with `clean_start = true` (Step 3, first run) to seed Bronze in + your bucket. Feldera's `snapshot_and_follow` needs the snapshot to exist before it + starts. +2. Start (or restart) the Feldera pipeline. It backfills the seeded snapshot, then begins + following the Delta log. +3. Run `ingest_bronze` for each later hour. As each run commits, Feldera ingests the change + and updates Gold sub-second. Watch a Gold view from the **Ad-hoc query** or + **Change stream** tab, exactly as in [Part 2](./part2.md#watch-a-gold-view-update-in-real-time). +4. Measure the latency. Run `measure_latency.py` against the pipeline to see, per Bronze + Delta connector, how long Feldera took to ingest and fully process the latest Delta + commit (see [Result: Feldera's latency](#result-feldera) below): + + ```bash + uv pip install feldera python-dotenv + # or + pip install feldera python-dotenv + ``` + + Then run: + + ```bash + uv run measure_latency.py --pipeline ecommerce-medallion-architecture + # or + python measure_latency.py --pipeline ecommerce-medallion-architecture + ``` + +## Notes {#notes} + +- **Run hours in order, once each.** The append-only tables use plain appends (the cheapest + Delta write, and what keeps them insert-only), so re-running an already-applied hour + double-inserts into them. `bronze_orders` is idempotent (MERGE). For replayable hours, + switch the append tables to an idempotent MERGE on their primary key or use a + `txnAppId`/`txnVersion` marker on the append. +- **`scale_factor` and `warehouse_bucket` are set in two places** — the job parameters + (drive the Bronze S3 path in `ingest_bronze.py`) and the pipeline `configuration` (drives + the path the materialized views read from). The job values override the pipeline's at run + time via matching keys; keep the defaults in sync. +- **Databricks incremental refresh is best-effort.** Databricks (Enzyme) incrementalizes a + materialized view when the query shape allows and otherwise falls back to a full + recompute — unlike Feldera, which maintains the entire DAG incrementally on every commit. + That difference is the point of the comparison. + +## Result: the Databricks refresh for the first CDC hour + +The numbers below are the `refresh_silver_gold` pipeline task rebuilding the Silver and Gold +materialized views defined in `silver_gold_pipeline.py` after the first CDC hour +(`2025-11-30T00`) lands in Bronze. Databricks reports each update as a sequence of phases; +these are the phases for that single refresh: + +| Phase | Time | What it is | +|---|---|---| +| Created | 1s | Update queued. | +| Waiting for resources | 8s | Serverless compute provisioning. | +| Initializing | 18s | Pipeline graph + environment setup. | +| Setting up tables | 1s | Reconciling materialized-view metadata. | +| **Running** | **58s** | **Refreshing the 15 Silver/Gold views.** | +| **Total** | **86s** | End-to-end, data-landed to Gold-current. | + +Read the phases as two distinct costs: + +- **Compute — 58s.** The actual work of refreshing the views for *one hour* of change. + Work is not proportional to the size of the change. The clean start with full recomputation took about as long as the incremental run. In addition, `gold_product_demand_surge` full + recomputes every run (see below), and every Lakeflow flow carries fixed per-update + overhead. This is the same SQL Feldera runs. +- **Freshness — 86s.** What a consumer waits from the moment Bronze receives the hour to + the moment Gold reflects it. Serverless re-provisions and re-initializes on every + triggered run, so the 28s of `Waiting for resources` + `Initializing` is paid again each + time. + +Feldera pays neither cost the same way. It is always on, so there is no per-run +provisioning, and it maintains every view incrementally, so the update is proportional to +the size of the change. Updates are reflected in less than a second for one CDC hour, not 58s. + +### The view Databricks cannot incrementalize + +`gold_product_demand_surge` compares each clickstream event against a trailing 24-hour +window measured from the latest event: + +```sql +WHERE ce.event_timestamp > + (SELECT MAX(event_timestamp) FROM silver_enriched_clickstream) - INTERVAL 1 DAY +``` + +Databricks' Enzyme incremental planner cannot handle a `SUBQUERY_EXPRESSION` inside a Filter/And operator. +Feldera maintains identical SQL, and any other SQL you might write incrementally. + +You can confirm which views incrementalized and which full recomputed in the pipeline's +**event log** (each flow reports its refresh type per update). + +## Result: Feldera's latency for the first CDC hour {#result-feldera} + +The same first CDC hour (`2025-11-30T00`), measured on the Feldera side. When +`ingest_bronze` commits the hour, Feldera follows the Delta log and propagates the change +through every Silver and Gold view. `measure_latency.py` reads the `completed_frontier` +timestamps each Delta connector exposes on `/stats` and reports three numbers per Bronze +table (see [measuring end-to-end latency with input frontiers](/pipelines/latency/#measuring-end-to-end-latency-with-input-frontiers)): + +- **ingest→proc** — from the change arriving at the connector to the IVM engine finishing it. +- **proc→done** — from processing to the outputs reaching all sinks. +- **e2e** — the full ingest-to-sink latency (their sum). + +`version` is the Bronze table's Delta version; version `1` is the first CDC commit on top +of the seeded snapshot: + +``` +table version ingest→proc proc→done e2e +---------------------------------------------------------------------- +bronze_clickstream_events 1 0.08s 0.05s 0.13s +bronze_inventory_events 1 0.08s 0.05s 0.13s +bronze_orders 1 0.29s 0.08s 0.37s +bronze_order_items 1 0.08s 0.07s 0.15s +``` + +Feldera reflected the first CDC hour end-to-end in **0.37s at most** (`bronze_orders`, whose +delete-and-insert updates cost more than the append-only streams) against Databricks' 86s +for the identical SQL. There is no per-run startup: Feldera is already running, so the only +cost is processing the change itself. + +## Feldera vs. the Databricks Lakeflow Declarative Pipelines + +| | Databricks workflow | Feldera | +|---|---|---| +| Silver / Gold | Materialized views, incremental *where possible* (14/15) | All 15 views, always incremental | +| `gold_product_demand_surge` | Full recompute every refresh | Incremental | +| Orchestration | Job schedule + pipeline refresh | None — change-driven | +| Per-run startup | ~28s serverless provisioning | None — always on | +| Freshness for the first CDC hour | ~86s end-to-end | 0.37s at most | + +Both consume the identical Bronze Delta tables in your bucket and run the identical Silver +and Gold SQL. Databricks maintains 14 of the 15 views incrementally and full recomputes +`gold_product_demand_surge` on every refresh, taking 86s end-to-end (58s of it compute) to +reflect the first CDC hour. Feldera follows the Delta log and maintains every view +incrementally, reflecting the same hour end-to-end in 0.37s at most, without any per-run +startup.