Build a Governed Telco Lakehouse Pipeline

Build a medallion architecture pipeline with quality gates on Databricks.

Introduction

30 Second Summary

Every time you make a phone call, send a text, or stream a video, your telecom provider generates a data record. Processing millions of these records daily while keeping the data accurate and governed is a core challenge for data engineering teams.

In this project, you will build a production-style data pipeline on Databricks that transforms raw telecommunications data through Bronze, Silver, and Gold layers. You will enforce data quality at every stage and visualize business metrics on a SQL dashboard.

What You'll Build

A governed telco lakehouse pipeline using Databricks Free Edition that ingests subscriber records, Call Detail Records, and plan data through a medallion architecture with declarative quality gates, Unity Catalog governance, workflow orchestration, and SQL analytics.

By the end of this project, you'll have:

  • A Unity Catalog schema structure with Bronze, Silver, and Gold layers enforcing data quality expectations at every stage.
  • A Lakeflow Spark Declarative Pipeline that rejects bad data before it reaches downstream consumers.
  • A scheduled Lakeflow Job and a Databricks SQL dashboard surfacing telco KPIs like ARPU, data usage trends, and subscriber segments.
  • Secret Mission: Implement SCD Type-2 and Type-4 patterns to track subscriber plan history and segment migrations over time.

Are there any prerequisites?

This project assumes familiarity with PySpark and SQL. You do not need a cloud account or credit card. You will sign up for Databricks Free Edition during the first step.

Set Up Your Databricks Workspace

Before you can build a pipeline, you need a workspace where your data lives and your code runs. Databricks Free Edition gives you a fully managed environment with serverless compute and Unity Catalog governance built in.

In this step, you will create your Databricks account, set up the medallion architecture schema structure (Bronze, Silver, Gold), and upload the sample telco data that your pipeline will process.

In this step, get ready to:
  • Sign up for Databricks Free Edition and access your workspace.
  • Create the Bronze, Silver, and Gold schemas under Unity Catalog.
  • Upload sample telco CSV files to a managed volume.
Sign up for Databricks Free Edition

Databricks Free Edition is completely free with no credit card and no cloud account required. It comes with serverless compute, Unity Catalog, and default storage ready to use.

  • Navigate to databricks.com/try-databricks in your browser.
  • Choose your preferred authentication method (email OTP, Sign in with Google, or Sign in with Microsoft).
  • Complete the signup flow.

Databricks creates a workspace for you automatically. Once you land on the workspace home page, you are ready to continue.

What does Free Edition include?

Your workspace comes with serverless compute for notebooks, one SQL warehouse, Unity Catalog for governance, and Lakeflow for pipelines and jobs.

Outbound internet access is restricted to trusted domains only. This means you cannot download external files from code. You will upload data manually through the workspace UI instead.

Create the Unity Catalog schema structure

The medallion architecture organizes data into three layers: Bronze (raw), Silver (cleansed), and Gold (business-ready). You will create a schema for each layer under your workspace catalog.

  • In the Databricks sidebar, click New, then click Notebook.
  • A blank notebook opens. Click the language dropdown at the top of the notebook and select SQL.
  • Name your notebook setup_schemas by clicking the title at the top and typing the name.

How do I find my workspace catalog name?

Click Catalog in the sidebar. Your workspace catalog shares a name with your workspace and appears at the top of the catalog panel. In the code below, replace workspace_catalog with your actual workspace catalog name.

  • Paste the following SQL into the first cell of your notebook, replacing workspace_catalog with your actual workspace catalog name:
-- Create medallion architecture schemas for telco data
USE CATALOG workspace_catalog;

CREATE SCHEMA IF NOT EXISTS bronze
COMMENT 'Raw ingested telco data - no transformations applied';

CREATE SCHEMA IF NOT EXISTS silver
COMMENT 'Cleansed, conformed, and deduplicated telco data';

CREATE SCHEMA IF NOT EXISTS gold
COMMENT 'Business-ready telco aggregations and KPI tables';

-- Create a managed volume for raw data upload
CREATE VOLUME IF NOT EXISTS bronze.raw_data;

What does this code do?

  • USE CATALOG sets the active catalog so all subsequent commands run within it.
  • Each CREATE SCHEMA statement creates a namespace for one medallion layer. The COMMENT makes the schema's purpose visible in Catalog Explorer.
  • CREATE VOLUME creates a managed storage location inside the bronze schema where you will upload raw CSV files.
  • Click Run All at the top of the notebook to execute every cell.
  • Verify each statement shows a green checkmark with no errors.
  • Click Catalog in the sidebar, then expand your workspace catalog. You should see three new schemas: bronze, silver, and gold.
  • Expand the bronze schema and confirm you see a raw_data volume listed.

Not seeing your schemas?

  • Make sure you replaced workspace_catalog with your actual catalog name. You can find it by clicking Catalog in the sidebar.
  • Check that the notebook language is set to SQL (not Python). The language dropdown is at the top of the notebook.
  • If you see a permissions error, ensure you are using the workspace catalog (the default one that shares your workspace name).

Help me troubleshoot why my schemas did not create in Databricks Free Edition.

✔️ Awesome, I've got everything!

Great. Make sure your notebook is saved, then continue to the next substep.

ⓧ I'd like to double check the full code

Here is the complete setup_schemas.sql notebook:

-- Create medallion architecture schemas for telco data
USE CATALOG workspace_catalog;

CREATE SCHEMA IF NOT EXISTS bronze
COMMENT 'Raw ingested telco data - no transformations applied';

CREATE SCHEMA IF NOT EXISTS silver
COMMENT 'Cleansed, conformed, and deduplicated telco data';

CREATE SCHEMA IF NOT EXISTS gold
COMMENT 'Business-ready telco aggregations and KPI tables';

-- Create a managed volume for raw data upload
CREATE VOLUME IF NOT EXISTS bronze.raw_data;
Upload the sample telco CSV files

Your pipeline needs raw data to process. You will create three CSV files on your local machine and upload them to the managed volume you just created.

Create a file called plans.csv on your computer with the following content:

plan_id,plan_name,plan_type,monthly_cost,data_limit_gb,voice_minutes_limit,sms_limit
PLAN001,Basic Voice,voice,29.99,2.0,500,100
PLAN002,Standard Mix,mixed,49.99,10.0,1000,500
PLAN003,Premium Data,data,89.99,50.0,2000,1000
PLAN004,Enterprise Plus,mixed,149.99,100.0,5000,2000

Create a file called subscribers.csv on your computer with the following content:

subscriber_id,name,email,phone_number,city,state,plan_id,activation_date,status,updated_at
SUB001,Alice Chen,alice.chen@email.com,555-0101,Austin,TX,PLAN003,2024-01-15,active,2025-06-01
SUB002,Bob Martinez,bob.m@email.com,555-0102,Denver,CO,PLAN002,2024-03-22,active,2025-05-15
SUB003,Carol Smith,carol.s@email.com,555-0103,Seattle,WA,PLAN001,2024-06-10,active,2025-04-20
SUB004,David Johnson,david.j@email.com,555-0104,Austin,TX,PLAN004,2023-11-01,active,2025-06-05
SUB005,Eva Williams,eva.w@email.com,555-0105,Portland,OR,PLAN002,2024-08-18,active,2025-03-10
SUB006,Frank Brown,frank.b@email.com,555-0106,Denver,CO,PLAN003,2024-02-28,suspended,2025-05-01
SUB007,Grace Lee,grace.lee@email.com,555-0107,Seattle,WA,PLAN001,2024-09-05,active,2025-06-08
SUB008,Henry Davis,henry.d@email.com,555-0108,Portland,OR,PLAN004,2023-12-15,active,2025-05-20
SUB009,Iris Patel,iris.p@email.com,555-0109,Austin,TX,PLAN002,2024-04-01,active,2025-04-15
SUB010,Jack Wilson,jack.w@email.com,555-0110,Denver,CO,PLAN003,2024-07-12,active,2025-06-10

Create a file called cdrs.csv on your computer with the following content:

cdr_id,subscriber_id,plan_id,call_timestamp,call_duration_seconds,data_usage_mb,sms_count,charge_amount,call_type
CDR001,SUB001,PLAN003,2025-06-01 08:15:00,180,250.5,3,185.00,data
CDR002,SUB001,PLAN003,2025-06-02 14:30:00,420,180.0,5,172.50,voice
CDR003,SUB002,PLAN002,2025-06-01 09:00:00,120,50.2,2,28.75,mixed
CDR004,SUB003,PLAN001,2025-06-01 10:45:00,60,10.0,1,8.25,voice
CDR005,SUB004,PLAN004,2025-06-01 11:20:00,600,500.0,10,185.00,data
CDR006,SUB004,PLAN004,2025-06-02 16:00:00,300,350.0,8,172.50,data
CDR007,SUB005,PLAN002,2025-06-01 13:00:00,90,25.0,4,22.00,mixed
CDR008,SUB006,PLAN003,2025-06-01 15:30:00,240,200.0,6,42.00,data
CDR009,SUB007,PLAN001,2025-06-02 08:00:00,45,5.0,0,3.50,voice
CDR010,SUB008,PLAN004,2025-06-01 19:00:00,480,450.0,12,285.00,data
CDR011,SUB008,PLAN004,2025-06-02 09:30:00,360,280.0,7,268.00,mixed
CDR012,SUB009,PLAN002,2025-06-01 12:15:00,150,75.0,3,18.50,mixed
CDR013,SUB010,PLAN003,2025-06-02 11:00:00,210,310.0,4,52.00,data
CDR014,SUB001,PLAN003,2025-06-03 10:00:00,300,220.0,6,178.00,mixed
CDR015,SUB002,PLAN002,2025-06-02 17:45:00,180,60.0,3,16.25,voice
CDR016,SUB004,PLAN004,2025-06-03 08:30:00,540,420.0,9,178.00,data
CDR017,SUB005,PLAN002,2025-06-02 14:00:00,75,30.0,2,10.50,voice
CDR018,SUB003,PLAN001,2025-06-02 16:30:00,30,8.0,1,4.75,voice
CDR019,SUB010,PLAN003,2025-06-03 13:15:00,270,290.0,5,48.00,data
CDR020,SUB009,PLAN002,2025-06-02 10:30:00,120,55.0,4,14.00,mixed

What is this data?

  • plans.csv contains the telco plan catalog (4 plans ranging from Basic Voice to Enterprise Plus).
  • subscribers.csv contains 10 subscriber records with plan assignments, locations, and activation dates.
  • cdrs.csv contains 20 Call Detail Records with timestamps, usage metrics, and charge amounts. These represent individual calls, data sessions, and SMS activity.

Now upload all three files to the raw_data volume in Databricks.

  • In the Databricks sidebar, click New, then click Add or upload data.
  • Click Upload files to a volume.
  • Under Files, click browse and select all three files (plans.csv, subscribers.csv, cdrs.csv). You can also drag and drop them into the upload area.
  • Under Destination volume, navigate to your workspace catalog, expand the bronze schema, and select the raw_data volume.
  • Wait for the upload to complete. You should see a success confirmation for each file.
  • Click Catalog in the sidebar. Navigate to your workspace catalog, expand bronze, then expand the raw_data volume. You should see subscribers.csv, cdrs.csv, and plans.csv listed inside.

Don't see your files in the volume?

  • Double-check that you selected the raw_data volume under the bronze schema as the destination. If you accidentally uploaded to a different location, re-upload to the correct volume.
  • Make sure the files have the .csv extension. Some text editors may add .txt by default.
  • Try refreshing the Catalog Explorer panel by clicking away and clicking Catalog again.

Help me troubleshoot why my uploaded CSV files are not appearing in the raw_data volume.

Why upload manually instead of downloading from code?

Databricks Free Edition restricts outbound internet access to a limited set of trusted domains. This means notebooks cannot download files from arbitrary URLs. Uploading through the workspace UI is the supported approach for getting local data into your volumes.

Your workspace is ready with a governed schema structure and raw telco data in place. Next up, you will build the Bronze and Silver pipeline that ingests this data and enforces quality at every layer.

Build the Bronze and Silver Pipeline

Your schemas are created and raw telco data is sitting in the volume, waiting to be processed. Now it's time to build the pipeline that transforms those CSV files into trusted, queryable Delta Lake tables.

In the medallion architecture, the Bronze layer ingests raw data exactly as-is, while the Silver layer cleanses, deduplicates, and conforms it. Data quality expectations act as gatekeepers at each layer, rejecting bad records before they pollute downstream tables.

In this step, get ready to:
  • Define Bronze streaming tables with data quality expectations that reject null IDs and invalid timestamps.
  • Build Silver materialized views that cast types, deduplicate records, and enforce referential integrity.
  • Create and trigger a Lakeflow Spark Declarative Pipeline to execute the notebook end-to-end.
Create the Bronze ingestion layer

The Bronze layer captures raw data with zero transformation. Its only job is to land data into Delta tables and reject records that are fundamentally broken (like missing primary keys).

  • In the Databricks sidebar, click + New and select Notebook.
  • Rename the notebook to telco_pipeline by clicking the title at the top.
  • Confirm the default language is set to Python in the language dropdown.
  • Clear any default content in the first cell, then paste the following code to define the imports and all three Bronze streaming tables:
from pyspark import pipelines as dp
from pyspark.sql.functions import col, sum, count, avg, when, round

# =============================================================================
# BRONZE LAYER - Raw Ingestion with Quality Gates
# =============================================================================

@dp.table(
    name="bronze_subscribers",
    comment="Raw subscriber records ingested from CSV"
)
@dp.expect_or_fail("valid_subscriber_id", "subscriber_id IS NOT NULL")
@dp.expect_or_fail("valid_phone", "phone_number IS NOT NULL AND length(phone_number) > 0")
def bronze_subscribers():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/subscribers")
        .load("/Volumes/workspace_catalog/bronze/raw_data/subscribers")
    )


@dp.table(
    name="bronze_cdrs",
    comment="Raw Call Detail Records ingested from CSV"
)
@dp.expect_or_fail("valid_cdr_id", "cdr_id IS NOT NULL")
@dp.expect_or_fail("valid_timestamp", "call_timestamp IS NOT NULL AND length(call_timestamp) > 0")
def bronze_cdrs():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/cdrs")
        .load("/Volumes/workspace_catalog/bronze/raw_data/cdrs")
    )


@dp.table(
    name="bronze_plans",
    comment="Raw plan catalog ingested from CSV"
)
@dp.expect_or_fail("valid_plan_id", "plan_id IS NOT NULL")
def bronze_plans():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/plans")
        .load("/Volumes/workspace_catalog/bronze/raw_data/plans")
    )

What does this code do?

from pyspark import pipelines as dp imports the Lakeflow Spark Declarative Pipelines module. This replaces the older import dlt syntax.

@dp.table declares a streaming table that incrementally ingests new files as they arrive. The cloudFiles format (Auto Loader) automatically detects and processes new CSV files from the volume path.

@dp.expect_or_fail defines a hard quality gate. If any record violates the constraint (like a null subscriber_id), the entire pipeline update fails. This prevents fundamentally broken data from entering the lakehouse.

Why Lakeflow SDP over plain notebooks?

You could write raw PySpark in a regular notebook and schedule it with a job. However, Lakeflow Spark Declarative Pipelines give you built-in data quality expectations, automatic dependency resolution between tables, and full lineage tracking in Unity Catalog.

With plain notebooks, you would need to manually implement quality checks, handle table ordering, and build your own lineage. SDP handles all of this declaratively.

You won't see output yet. This code defines the pipeline structure, but it only executes when you trigger a pipeline update in the final substep.

Define the Silver transformation layer

The Silver layer reads from Bronze, applies type casting, removes duplicates, and enforces business rules. Records that fail Silver expectations get dropped silently instead of crashing the pipeline.

  • In the same notebook cell, paste the following Silver layer code directly below the Bronze code you just added:
# =============================================================================
# SILVER LAYER - Cleansed and Conformed
# =============================================================================

silver_cdr_rules = {
    "valid_subscriber_ref": "subscriber_id IS NOT NULL",
    "valid_plan_ref": "plan_id IS NOT NULL",
    "non_negative_duration": "call_duration_seconds >= 0",
    "non_negative_data": "data_usage_mb >= 0"
}

@dp.materialized_view(
    name="silver_cdrs",
    comment="Cleansed CDRs with type casting and quality enforcement"
)
@dp.expect_all_or_drop(silver_cdr_rules)
def silver_cdrs():
    return (
        spark.read.table("bronze_cdrs")
        .select(
            col("cdr_id").cast("string"),
            col("subscriber_id").cast("string"),
            col("plan_id").cast("string"),
            col("call_timestamp").cast("timestamp"),
            col("call_duration_seconds").cast("int"),
            col("data_usage_mb").cast("decimal(10,2)"),
            col("sms_count").cast("int"),
            col("charge_amount").cast("decimal(10,2)"),
            col("call_type").cast("string")
        )
        .dropDuplicates(["cdr_id"])
    )


@dp.materialized_view(
    name="silver_subscribers",
    comment="Cleansed subscriber data with standardized fields"
)
@dp.expect_or_drop("valid_email", "email IS NOT NULL AND email LIKE '%@%'")
def silver_subscribers():
    return (
        spark.read.table("bronze_subscribers")
        .select(
            col("subscriber_id").cast("string"),
            col("name").cast("string"),
            col("email").cast("string"),
            col("phone_number").cast("string"),
            col("city").cast("string"),
            col("state").cast("string"),
            col("plan_id").cast("string"),
            col("activation_date").cast("date"),
            col("status").cast("string")
        )
        .dropDuplicates(["subscriber_id"])
    )


@dp.materialized_view(
    name="silver_plans",
    comment="Cleansed plan catalog"
)
@dp.expect_or_drop("valid_monthly_cost", "monthly_cost > 0")
def silver_plans():
    return (
        spark.read.table("bronze_plans")
        .select(
            col("plan_id").cast("string"),
            col("plan_name").cast("string"),
            col("plan_type").cast("string"),
            col("monthly_cost").cast("decimal(10,2)"),
            col("data_limit_gb").cast("decimal(5,1)"),
            col("voice_minutes_limit").cast("int"),
            col("sms_limit").cast("int")
        )
        .dropDuplicates(["plan_id"])
    )

What does this code do?

@dp.materialized_view creates a batch-computed view that reads from Bronze and materializes the results as a Delta table. Unlike streaming tables, materialized views recompute fully on each refresh.

silver_cdr_rules is a dictionary of constraints passed to @dp.expect_all_or_drop. Any CDR record that violates any rule (null subscriber reference, negative duration, negative data usage) gets silently dropped instead of failing the pipeline.

The .dropDuplicates() call removes exact duplicate records by primary key, while .cast() enforces proper data types (timestamps, decimals, integers) on columns that arrived as raw strings from CSV.

What is the difference between expect_or_fail and expect_or_drop?

@dp.expect_or_fail stops the entire pipeline if a violation is found. Use it for Bronze-layer integrity where bad records signal a source problem that needs investigation.

@dp.expect_or_drop silently removes violating records and lets the pipeline continue. Use it for Silver-layer business rules where occasional bad records are expected and should not block downstream processing.

Your notebook now defines the full Bronze-to-Silver pipeline. The next substep will execute it.

✔️ Awesome, I've got everything!

Great. Double check your notebook is saved (Databricks auto-saves, but you can press Cmd+S or Ctrl+S to be safe).

ⓧ I'd like to double check the full code

Here is the complete notebook code you should have at this point:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, sum, count, avg, when, round

# =============================================================================
# BRONZE LAYER - Raw Ingestion with Quality Gates
# =============================================================================

@dp.table(
    name="bronze_subscribers",
    comment="Raw subscriber records ingested from CSV"
)
@dp.expect_or_fail("valid_subscriber_id", "subscriber_id IS NOT NULL")
@dp.expect_or_fail("valid_phone", "phone_number IS NOT NULL AND length(phone_number) > 0")
def bronze_subscribers():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/subscribers")
        .load("/Volumes/workspace_catalog/bronze/raw_data/subscribers")
    )


@dp.table(
    name="bronze_cdrs",
    comment="Raw Call Detail Records ingested from CSV"
)
@dp.expect_or_fail("valid_cdr_id", "cdr_id IS NOT NULL")
@dp.expect_or_fail("valid_timestamp", "call_timestamp IS NOT NULL AND length(call_timestamp) > 0")
def bronze_cdrs():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/cdrs")
        .load("/Volumes/workspace_catalog/bronze/raw_data/cdrs")
    )


@dp.table(
    name="bronze_plans",
    comment="Raw plan catalog ingested from CSV"
)
@dp.expect_or_fail("valid_plan_id", "plan_id IS NOT NULL")
def bronze_plans():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/plans")
        .load("/Volumes/workspace_catalog/bronze/raw_data/plans")
    )


# =============================================================================
# SILVER LAYER - Cleansed and Conformed
# =============================================================================

silver_cdr_rules = {
    "valid_subscriber_ref": "subscriber_id IS NOT NULL",
    "valid_plan_ref": "plan_id IS NOT NULL",
    "non_negative_duration": "call_duration_seconds >= 0",
    "non_negative_data": "data_usage_mb >= 0"
}

@dp.materialized_view(
    name="silver_cdrs",
    comment="Cleansed CDRs with type casting and quality enforcement"
)
@dp.expect_all_or_drop(silver_cdr_rules)
def silver_cdrs():
    return (
        spark.read.table("bronze_cdrs")
        .select(
            col("cdr_id").cast("string"),
            col("subscriber_id").cast("string"),
            col("plan_id").cast("string"),
            col("call_timestamp").cast("timestamp"),
            col("call_duration_seconds").cast("int"),
            col("data_usage_mb").cast("decimal(10,2)"),
            col("sms_count").cast("int"),
            col("charge_amount").cast("decimal(10,2)"),
            col("call_type").cast("string")
        )
        .dropDuplicates(["cdr_id"])
    )


@dp.materialized_view(
    name="silver_subscribers",
    comment="Cleansed subscriber data with standardized fields"
)
@dp.expect_or_drop("valid_email", "email IS NOT NULL AND email LIKE '%@%'")
def silver_subscribers():
    return (
        spark.read.table("bronze_subscribers")
        .select(
            col("subscriber_id").cast("string"),
            col("name").cast("string"),
            col("email").cast("string"),
            col("phone_number").cast("string"),
            col("city").cast("string"),
            col("state").cast("string"),
            col("plan_id").cast("string"),
            col("activation_date").cast("date"),
            col("status").cast("string")
        )
        .dropDuplicates(["subscriber_id"])
    )


@dp.materialized_view(
    name="silver_plans",
    comment="Cleansed plan catalog"
)
@dp.expect_or_drop("valid_monthly_cost", "monthly_cost > 0")
def silver_plans():
    return (
        spark.read.table("bronze_plans")
        .select(
            col("plan_id").cast("string"),
            col("plan_name").cast("string"),
            col("plan_type").cast("string"),
            col("monthly_cost").cast("decimal(10,2)"),
            col("data_limit_gb").cast("decimal(5,1)"),
            col("voice_minutes_limit").cast("int"),
            col("sms_limit").cast("int")
        )
        .dropDuplicates(["plan_id"])
    )
Configure and run the pipeline

The notebook defines what the pipeline does, but it needs a pipeline resource to actually execute it. You will now create that resource and trigger your first pipeline update.

  • In the Databricks sidebar, click Jobs & Pipelines.
  • Click New, then select ETL Pipeline.
  • Name your pipeline Telco Lakehouse Pipeline by clicking the title area at the top.
  • Select Add existing assets as your creation method.
  • In the asset browser, navigate to and select your telco_pipeline notebook as the source.
  • Under the pipeline name, confirm the default catalog is set to your workspace catalog and the default schema is set to bronze.

Why set the default schema to bronze?

The default catalog and schema tell the pipeline where to publish tables when your code does not explicitly qualify a table name with a catalog or schema prefix. Since your code uses unqualified names like bronze_cdrs and silver_cdrs, the pipeline publishes all tables into this default location.

  • Click Run pipeline in the upper right corner to trigger a pipeline update.

The pipeline will take a few minutes to provision serverless compute, process the CSV files through Bronze, and materialize the Silver tables. You can watch progress in the pipeline DAG view.

  • Wait for the pipeline update to complete. You should see all six tables (three Bronze, three Silver) displayed in the DAG with green expectation badges indicating all quality checks passed.

Pipeline update failed?

  • Check that your CSV files were uploaded to the exact path /Volumes/workspace_catalog/bronze/raw_data/. The pipeline cannot read files from a different location.
  • Verify that your notebook language is Python, not SQL. A SQL notebook will not recognize the @dp.table decorators.
  • If you see expectation violations (red badges), your CSV data contains null values in required columns. Check that your CSV files have proper headers and non-empty ID fields.

Help me debug my Lakeflow pipeline update failure

Your Bronze and Silver pipeline is running. Raw telco data flows in, gets validated at the Bronze gate, then gets cleansed and conformed in Silver. Next up, you will build the Gold layer that transforms this clean data into business KPIs like ARPU and subscriber segments.

Build the Gold Layer Aggregations

Your Bronze and Silver layers are ingesting and cleansing telco data. Now it's time to build the layer that business stakeholders actually consume.

The Gold layer transforms cleansed data into business-ready metrics. In this step, you'll create materialized views that calculate ARPU (Average Revenue Per User), track daily usage trends, and segment subscribers into tiers based on their spending patterns.

In this step, get ready to:
  • Define Gold materialized views for telco KPIs (ARPU, usage trends, subscriber segments).
  • Enforce data quality expectations at the Gold layer.
  • Verify end-to-end pipeline lineage from Bronze through Silver to Gold.
Define Gold materialized views

Gold views aggregate Silver data into metrics that answer business questions directly. Each view reads from the cleansed Silver tables and produces a focused output optimized for analytics.

  • Switch back to your telco_pipeline.py notebook in the Databricks workspace.
  • Scroll to the bottom of the file (after the Silver layer code) and add the Gold layer section header, the ARPU rules dictionary, and the monthly_arpu view by pasting this code:
# =============================================================================
# GOLD LAYER - Telco Business Aggregations
# =============================================================================

gold_arpu_rules = {
    "non_negative_arpu": "avg_revenue_per_user >= 0",
    "valid_billing_month": "billing_month IS NOT NULL"
}

@dp.materialized_view(
    name="monthly_arpu",
    comment="Average Revenue Per User by billing month"
)
@dp.expect_all_or_fail(gold_arpu_rules)
def monthly_arpu():
    return (
        spark.read.table("silver_cdrs")
        .withColumn("billing_month", col("call_timestamp").cast("date").substr(1, 7))
        .groupBy("billing_month")
        .agg(
            round(avg("charge_amount"), 2).alias("avg_revenue_per_user"),
            sum("charge_amount").alias("total_revenue"),
            count("subscriber_id").alias("total_transactions")
        )
    )

What does this code do?

The gold_arpu_rules dictionary defines two constraints that MUST pass for every row. If any row has a negative ARPU or a null billing month, the entire pipeline update fails. This protects downstream dashboards from displaying misleading data.

The monthly_arpu view reads from silver_cdrs, extracts the year-month from each call timestamp, then groups by that month to calculate average revenue per user, total revenue, and transaction counts.

  • Below the monthly_arpu function, add the usage_trends view by pasting this code:

@dp.materialized_view(
    name="usage_trends",
    comment="Daily data and voice usage trends"
)
@dp.expect_or_fail("valid_usage_date", "usage_date IS NOT NULL")
def usage_trends():
    return (
        spark.read.table("silver_cdrs")
        .withColumn("usage_date", col("call_timestamp").cast("date"))
        .groupBy("usage_date")
        .agg(
            round(sum("data_usage_mb"), 2).alias("total_data_mb"),
            sum("call_duration_seconds").alias("total_voice_seconds"),
            sum("sms_count").alias("total_sms"),
            count("cdr_id").alias("total_records")
        )
    )

What does this view calculate?

This view groups CDR data by calendar date and sums up data consumption (MB), voice duration (seconds), SMS count, and total records per day. The output powers a time-series chart showing how subscriber usage patterns shift over time.

  • Below the usage_trends function, add the subscriber_segments view by pasting this code:

@dp.materialized_view(
    name="subscriber_segments",
    comment="Subscriber segmentation by monthly spend and usage"
)
@dp.expect_or_fail("valid_subscriber", "subscriber_id IS NOT NULL")
def subscriber_segments():
    return (
        spark.read.table("silver_cdrs")
        .join(spark.read.table("silver_subscribers"), "subscriber_id")
        .join(spark.read.table("silver_plans"), "plan_id")
        .groupBy("subscriber_id", "name", "phone_number", "plan_name", "plan_type", "city", "state")
        .agg(
            sum("charge_amount").alias("total_spend"),
            sum("data_usage_mb").alias("total_data_mb"),
            sum("call_duration_seconds").alias("total_voice_seconds"),
            count("cdr_id").alias("total_interactions")
        )
        .withColumn(
            "segment",
            when(col("total_spend") >= 500, "Premium")
            .when(col("total_spend") >= 100, "Standard")
            .otherwise("Basic")
        )
    )

What does the segmentation logic do?

This view joins three Silver tables (CDRs, subscribers, and plans) to build a full profile of each subscriber. It aggregates their total spend, data usage, and voice consumption.

The when column at the end assigns each subscriber to a tier. Subscribers who spend $500 or more are Premium. Those at $100 or more are Standard. Everyone else is Basic.

  • Save the telco_pipeline.py notebook.
  • You won't see changes yet. The next substep triggers a pipeline refresh that processes all three Gold views at once.

Seeing a syntax error in the notebook?

Check that each function definition starts with def and ends with a properly closed parenthesis. A missing closing ) on the return statement is the most common issue with chained PySpark transformations.

Also confirm the gold_arpu_rules dictionary is defined ABOVE the monthly_arpu function. Python requires variables to exist before they're referenced as decorator arguments.

help me debug a syntax error in my Databricks pipeline notebook

✔️ Awesome, I've got everything!

Great. Double check you've saved your notebook before moving on.

ⓧ I'd like to double check the full code

from pyspark import pipelines as dp
from pyspark.sql.functions import col, sum, count, avg, when, round

# =============================================================================
# BRONZE LAYER - Raw Ingestion with Quality Gates
# =============================================================================

@dp.table(
    name="bronze_subscribers",
    comment="Raw subscriber records ingested from CSV"
)
@dp.expect_or_fail("valid_subscriber_id", "subscriber_id IS NOT NULL")
@dp.expect_or_fail("valid_phone", "phone_number IS NOT NULL AND length(phone_number) > 0")
def bronze_subscribers():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/subscribers")
        .load("/Volumes/workspace_catalog/bronze/raw_data/subscribers")
    )


@dp.table(
    name="bronze_cdrs",
    comment="Raw Call Detail Records ingested from CSV"
)
@dp.expect_or_fail("valid_cdr_id", "cdr_id IS NOT NULL")
@dp.expect_or_fail("valid_timestamp", "call_timestamp IS NOT NULL AND length(call_timestamp) > 0")
def bronze_cdrs():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/cdrs")
        .load("/Volumes/workspace_catalog/bronze/raw_data/cdrs")
    )


@dp.table(
    name="bronze_plans",
    comment="Raw plan catalog ingested from CSV"
)
@dp.expect_or_fail("valid_plan_id", "plan_id IS NOT NULL")
def bronze_plans():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("header", "true")
        .option("cloudFiles.schemaLocation", "/tmp/schema/plans")
        .load("/Volumes/workspace_catalog/bronze/raw_data/plans")
    )


# =============================================================================
# SILVER LAYER - Cleansed and Conformed
# =============================================================================

silver_cdr_rules = {
    "valid_subscriber_ref": "subscriber_id IS NOT NULL",
    "valid_plan_ref": "plan_id IS NOT NULL",
    "non_negative_duration": "call_duration_seconds >= 0",
    "non_negative_data": "data_usage_mb >= 0"
}

@dp.materialized_view(
    name="silver_cdrs",
    comment="Cleansed CDRs with type casting and quality enforcement"
)
@dp.expect_all_or_drop(silver_cdr_rules)
def silver_cdrs():
    return (
        spark.read.table("bronze_cdrs")
        .select(
            col("cdr_id").cast("string"),
            col("subscriber_id").cast("string"),
            col("plan_id").cast("string"),
            col("call_timestamp").cast("timestamp"),
            col("call_duration_seconds").cast("int"),
            col("data_usage_mb").cast("decimal(10,2)"),
            col("sms_count").cast("int"),
            col("charge_amount").cast("decimal(10,2)"),
            col("call_type").cast("string")
        )
        .dropDuplicates(["cdr_id"])
    )


@dp.materialized_view(
    name="silver_subscribers",
    comment="Cleansed subscriber data with standardized fields"
)
@dp.expect_or_drop("valid_email", "email IS NOT NULL AND email LIKE '%@%'")
def silver_subscribers():
    return (
        spark.read.table("bronze_subscribers")
        .select(
            col("subscriber_id").cast("string"),
            col("name").cast("string"),
            col("email").cast("string"),
            col("phone_number").cast("string"),
            col("city").cast("string"),
            col("state").cast("string"),
            col("plan_id").cast("string"),
            col("activation_date").cast("date"),
            col("status").cast("string")
        )
        .dropDuplicates(["subscriber_id"])
    )


@dp.materialized_view(
    name="silver_plans",
    comment="Cleansed plan catalog"
)
@dp.expect_or_drop("valid_monthly_cost", "monthly_cost > 0")
def silver_plans():
    return (
        spark.read.table("bronze_plans")
        .select(
            col("plan_id").cast("string"),
            col("plan_name").cast("string"),
            col("plan_type").cast("string"),
            col("monthly_cost").cast("decimal(10,2)"),
            col("data_limit_gb").cast("decimal(5,1)"),
            col("voice_minutes_limit").cast("int"),
            col("sms_limit").cast("int")
        )
        .dropDuplicates(["plan_id"])
    )


# =============================================================================
# GOLD LAYER - Telco Business Aggregations
# =============================================================================

gold_arpu_rules = {
    "non_negative_arpu": "avg_revenue_per_user >= 0",
    "valid_billing_month": "billing_month IS NOT NULL"
}

@dp.materialized_view(
    name="monthly_arpu",
    comment="Average Revenue Per User by billing month"
)
@dp.expect_all_or_fail(gold_arpu_rules)
def monthly_arpu():
    return (
        spark.read.table("silver_cdrs")
        .withColumn("billing_month", col("call_timestamp").cast("date").substr(1, 7))
        .groupBy("billing_month")
        .agg(
            round(avg("charge_amount"), 2).alias("avg_revenue_per_user"),
            sum("charge_amount").alias("total_revenue"),
            count("subscriber_id").alias("total_transactions")
        )
    )


@dp.materialized_view(
    name="usage_trends",
    comment="Daily data and voice usage trends"
)
@dp.expect_or_fail("valid_usage_date", "usage_date IS NOT NULL")
def usage_trends():
    return (
        spark.read.table("silver_cdrs")
        .withColumn("usage_date", col("call_timestamp").cast("date"))
        .groupBy("usage_date")
        .agg(
            round(sum("data_usage_mb"), 2).alias("total_data_mb"),
            sum("call_duration_seconds").alias("total_voice_seconds"),
            sum("sms_count").alias("total_sms"),
            count("cdr_id").alias("total_records")
        )
    )


@dp.materialized_view(
    name="subscriber_segments",
    comment="Subscriber segmentation by monthly spend and usage"
)
@dp.expect_or_fail("valid_subscriber", "subscriber_id IS NOT NULL")
def subscriber_segments():
    return (
        spark.read.table("silver_cdrs")
        .join(spark.read.table("silver_subscribers"), "subscriber_id")
        .join(spark.read.table("silver_plans"), "plan_id")
        .groupBy("subscriber_id", "name", "phone_number", "plan_name", "plan_type", "city", "state")
        .agg(
            sum("charge_amount").alias("total_spend"),
            sum("data_usage_mb").alias("total_data_mb"),
            sum("call_duration_seconds").alias("total_voice_seconds"),
            count("cdr_id").alias("total_interactions")
        )
        .withColumn(
            "segment",
            when(col("total_spend") >= 500, "Premium")
            .when(col("total_spend") >= 100, "Standard")
            .otherwise("Basic")
        )
    )
Understand the Gold layer quality gates

You've already included data quality expectations as decorators on each Gold view. Let's understand the two patterns you used and why they matter at the Gold layer.

How do Gold layer expectations differ from Silver?

In the Silver layer, you used @dp.expect_all_or_drop and @dp.expect_or_drop. These silently drop rows that fail validation. That makes sense for cleaning raw data where some records are expected to be malformed.

In the Gold layer, you used @dp.expect_all_or_fail and @dp.expect_or_fail. These halt the entire pipeline update if any row violates the constraint. At the Gold layer, bad data means your aggregation logic has a bug. A failing pipeline is better than a dashboard showing negative ARPU to executives.

The gold_arpu_rules dictionary groups multiple constraints into a single decorator call. This keeps related rules together and makes them easy to maintain as business logic evolves.

Why use a dictionary for expectations?

When a view has multiple constraints, the dictionary pattern (@dp.expect_all_or_fail(dict)) is cleaner than stacking individual @dp.expect_or_fail decorators. Each key is a human-readable description that appears in the pipeline DAG alongside the expectation badge.

For views with just one constraint (like usage_trends and subscriber_segments), a single @dp.expect_or_fail decorator is simpler and equally effective.

Trigger a full pipeline refresh and verify lineage

With the Gold views defined, you need to run the pipeline again so Databricks processes the new materialized views. This time, the pipeline will refresh all three layers end-to-end.

  • In the left sidebar, click Jobs and Pipelines.
  • Click your existing Lakeflow Spark Declarative Pipeline (the one you created in the previous step).
  • Click Start to trigger a full pipeline refresh.
  • Wait for the pipeline DAG to update. You should see the three new Gold tables (monthly_arpu, usage_trends, subscriber_segments) appear with green expectation badges.

What should I see?

The pipeline DAG now shows your full medallion architecture. Bronze streaming tables feed into Silver materialized views, which feed into Gold materialized views. Each node displays a green badge indicating all expectations passed.

If you see a red badge or a failed status, click the node to inspect which expectation was violated. This usually means the source data has an unexpected null or negative value that slipped through Silver.

  • In the left sidebar, click Catalog.
  • Expand your workspace catalog, then expand the gold schema.
  • Click on monthly_arpu (or any Gold table).
  • Click the Lineage tab.
  • Click See Lineage Graph to open the interactive visualization.
  • Confirm you can trace the data flow from Bronze tables through Silver views to the Gold view.

What is data lineage?

Data lineage tracks where data comes from, how it was transformed, and where it flows next. Unity Catalog captures this automatically for every table and view in your pipeline. You never have to configure it manually.

The lineage graph is invaluable for debugging. If a Gold metric looks wrong, you can trace it back through Silver to Bronze to identify exactly where the problem was introduced.

Don't see the Gold tables in the pipeline DAG?

Make sure you saved the notebook before triggering the pipeline refresh. The pipeline reads the notebook source at trigger time. Any unsaved changes are invisible to the pipeline engine.

If the pipeline shows an error on subscriber_segments, check that your Silver tables from the previous step still exist. The three-way join requires silver_cdrs, silver_subscribers, and silver_plans to be populated.

help me troubleshoot my Gold layer pipeline refresh failing in Databricks

Your pipeline now transforms raw telco data all the way from ingestion to business metrics. Next up, you'll build a SQL dashboard that visualizes these Gold layer KPIs for stakeholders.

Create a Databricks SQL Dashboard

Your Gold layer tables are populated with telco business metrics. The next step is making those insights accessible to stakeholders who need answers without writing SQL themselves.

In this step, you will write analytical queries against the Gold layer and build a Databricks SQL dashboard that visualizes ARPU trends, usage patterns, and subscriber segmentation.

In this step, get ready to:
  • Write SQL queries against the Gold layer tables.
  • Build an AI/BI dashboard with four visualization panels.
Write the ARPU trend query

Average Revenue Per User (ARPU) is one of the most important KPIs in telecommunications. Tracking it over time reveals whether your subscriber base is growing in value or declining.

  • Click SQL Editor in the sidebar to open the query editor.
  • Click the + tab icon, then select Create new query to start a fresh query tab.
  • Paste the following ARPU trend query:
-- ARPU Trend (Line Chart)
SELECT
    billing_month,
    avg_revenue_per_user,
    total_revenue
FROM gold.monthly_arpu
ORDER BY billing_month;

What does this query do?

This query pulls the pre-aggregated ARPU values from your Gold layer monthly_arpu table. It returns the billing month, average revenue per user, and total revenue sorted chronologically.

Because the Gold layer already performed the heavy aggregation in the pipeline, this query is lightweight. It reads finished metrics rather than recomputing them from raw CDRs.

  • Press Cmd+Enter (macOS) or Ctrl+Enter (Windows) to run the query.
  • You should see a results table with rows showing each billing month alongside an avg_revenue_per_user and total_revenue value.

Seeing an empty result set?

Confirm your SQL warehouse is running. Check the compute selector at the top of the editor and verify a warehouse is connected.

If you see a table-not-found error, verify the schema path. Your Gold tables live under your workspace catalog, so the full path is workspace_catalog.gold.monthly_arpu. If your catalog and schema selectors are already set to the workspace catalog and gold schema, the short name gold.monthly_arpu works.

Help me troubleshoot my Gold layer query

Write the usage trends query

Usage trends show daily consumption of data, voice, and SMS across all subscribers. This is the foundation for capacity planning and identifying peak-usage periods.

  • Below your ARPU query (after the semicolon), paste the daily usage trends query:
-- Daily Usage Trends (Stacked Area Chart)
SELECT
    usage_date,
    total_data_mb,
    total_voice_seconds,
    total_sms
FROM gold.usage_trends
ORDER BY usage_date;

What does this query do?

This pulls daily aggregated usage from your usage_trends Gold table. Each row represents one day with total data consumption in MB, total voice duration in seconds, and total SMS count.

Stacking these three metrics on a chart reveals the relative contribution of each service type over time.

  • Highlight this query and press Cmd+Enter (macOS) or Ctrl+Enter (Windows) to run it.
  • You should see rows with usage_date, total_data_mb, total_voice_seconds, and total_sms columns populated.
Write the segment distribution and summary KPI queries

Subscriber segmentation groups your customer base into tiers based on spending. Summary KPIs give executives a single-glance view of overall performance.

  • Below your usage trends query, paste the subscriber segment distribution query:
-- Subscriber Segment Distribution (Pie Chart)
SELECT
    segment,
    COUNT(*) AS subscriber_count,
    SUM(total_spend) AS segment_revenue
FROM gold.subscriber_segments
GROUP BY segment
ORDER BY segment_revenue DESC;
  • Below that, paste the summary KPI counters query:
-- Summary KPIs (Counters)
SELECT
    SUM(total_revenue) AS total_revenue,
    AVG(avg_revenue_per_user) AS overall_arpu,
    SUM(total_transactions) AS total_transactions
FROM gold.monthly_arpu;

What do these queries do?

The segment distribution query counts subscribers in each tier (Premium, Standard, Basic) and sums their total spend. This powers a pie chart showing revenue concentration across tiers.

The summary KPI query calculates headline numbers: total revenue across all months, overall ARPU, and total transaction count. These power counter widgets that give an at-a-glance summary of the business.

  • Highlight each query and run them individually to verify results.
  • The segment query should return three rows (Premium, Standard, Basic). The KPI query should return a single row with three aggregated values.

Segment query returning zero rows?

The subscriber_segments view joins three Silver tables. If any Silver table is empty (due to a failed pipeline run), this query returns nothing. Return to your pipeline and verify all tables show green expectation badges.

Help me debug my subscriber segments query

✔️ Awesome, I've got everything!

Great. Make sure all four queries are saved in the same query tab. You will reference them as datasets in the dashboard next.

ⓧ I'd like to double check the full code

Here is the complete set of dashboard queries:

-- ARPU Trend (Line Chart)
SELECT
    billing_month,
    avg_revenue_per_user,
    total_revenue
FROM gold.monthly_arpu
ORDER BY billing_month;

-- Daily Usage Trends (Stacked Area Chart)
SELECT
    usage_date,
    total_data_mb,
    total_voice_seconds,
    total_sms
FROM gold.usage_trends
ORDER BY usage_date;

-- Subscriber Segment Distribution (Pie Chart)
SELECT
    segment,
    COUNT(*) AS subscriber_count,
    SUM(total_spend) AS segment_revenue
FROM gold.subscriber_segments
GROUP BY segment
ORDER BY segment_revenue DESC;

-- Summary KPIs (Counters)
SELECT
    SUM(total_revenue) AS total_revenue,
    AVG(avg_revenue_per_user) AS overall_arpu,
    SUM(total_transactions) AS total_transactions
FROM gold.monthly_arpu;
Build the dashboard with visualization panels

Now you will turn those query results into an interactive dashboard. Databricks AI/BI dashboards separate data definitions (the Data tab) from visual layout (the Canvas tab).

  • Click + New in the sidebar and select Dashboard.
  • A new blank dashboard opens with a Data tab and a Canvas tab at the top.

What is an AI/BI dashboard?

AI/BI dashboards are Databricks' interactive visualization platform. You define datasets (SQL queries that pull from your Unity Catalog tables), then build visualizations on a canvas that automatically refreshes when the underlying data changes.

Datasets inherit Unity Catalog permissions. Anyone you share the dashboard with can only see data they have access to.

Start by adding your queries as datasets:

  • Click the Data tab.
  • Click Create from SQL to add your first dataset.
  • Paste the ARPU trend query (the first SELECT from your queries) and click Run to produce the data collection.
  • Name this dataset ARPU Trend.
  • Repeat for the remaining three queries, naming them Usage Trends, Segment Distribution, and Summary KPIs.

Now build the visualization panels:

  • Switch to the Canvas tab.
  • Click Add a visualization and place it on the canvas.
  • In the configuration panel on the right, select the ARPU Trend dataset from the Dataset dropdown.
  • From the Visualization dropdown, select Line.
  • Set the X axis to billing_month and the Y axis to avg_revenue_per_user.
  • Add a second visualization. Select the Usage Trends dataset.
  • From the Visualization dropdown, select Area.
  • Set the X axis to usage_date.
  • Add total_data_mb and total_voice_seconds to the Y axis.
  • Click the kebab menu in the Y Axis section and choose Stack under Layout to create a stacked area chart.
  • Add a third visualization. Select the Segment Distribution dataset.
  • From the Visualization dropdown, select Pie.
  • Set the dimension to segment and the value to subscriber_count.
  • Add a fourth visualization. Select the Summary KPIs dataset.
  • From the Visualization dropdown, select Counter.
  • Set the value field to total_revenue. You can clone this widget and change the value to overall_arpu or total_transactions for additional counter tiles.
  • Drag the visualization widgets to arrange them on the canvas. Place the line chart and area chart side by side at the top, and the pie chart and counters below.

You should now see four panels rendering live data from your Gold layer tables.

Why build a dashboard instead of just running queries?

Queries give you answers. Dashboards give stakeholders answers without requiring them to write SQL. A published dashboard refreshes automatically and can be shared with anyone in your Databricks account.

In production, these dashboards are often the interface between the data engineering team (who builds the pipeline) and the business users (who make decisions based on the metrics).

Visualization showing no data?

Confirm you selected the correct dataset in the configuration panel. Each visualization must point to the dataset that contains its expected columns.

Also verify that a SQL warehouse is attached. The dashboard uses the SQL warehouse to execute its dataset queries. In Free Edition, you have one SQL warehouse available.

Help me fix my dashboard visualization

Your telco analytics are now visualized and ready for stakeholders. Next up, you will orchestrate the entire pipeline with a scheduled Lakeflow Job so this dashboard always reflects fresh data.

Orchestrate with a Scheduled Lakeflow Job

Your pipeline transforms raw telco data through Bronze, Silver, and Gold layers. Your dashboard surfaces the resulting KPIs. The missing piece is automation.

Right now, refreshing the pipeline requires a manual click. In production, data arrives continuously and teams expect fresh metrics without human intervention. A Lakeflow Job solves this by wrapping your pipeline in a scheduled, orchestrated workflow that runs end-to-end on its own.

In this step, get ready to:
  • Create a Lakeflow Job with a Pipeline task that triggers a full pipeline refresh.
  • Configure a scheduled trigger to run the job once per day.
  • Validate end-to-end orchestration by running the job manually and inspecting the results.
Create a Lakeflow Job with a Pipeline task

A Pipeline task is a special job task type that triggers a full refresh of your Lakeflow Spark Declarative Pipeline. One task covers the entire Bronze-through-Gold flow as a single orchestrated unit.

  • In the workspace sidebar, click Jobs & Pipelines.
  • Click Create, then select Job.
  • You will see a task configuration panel. Click Add another task type and search for Pipeline.
  • In the Task name field, enter a name such as refresh-telco-pipeline.
  • In the Pipeline drop-down menu, select your existing Lakeflow Spark Declarative Pipeline (the one running telco_pipeline.py).
  • Click Create task.
  • Replace the default job name (New Job) at the top of the page with something descriptive, such as telco-pipeline-daily.

What does this configuration do?

A Pipeline task tells the job to trigger a full update of your Lakeflow Spark Declarative Pipeline. The pipeline itself manages compute and execution order. All you need is one task that points to the pipeline, and the entire Bronze-through-Gold refresh runs as a single coordinated operation.

You should now see your job with a single Pipeline task displayed in the task graph.

Don't see your pipeline in the drop-down?

Only triggered pipelines appear in the Pipeline drop-down. Continuous pipelines are not supported as a job task. If your pipeline does not appear, confirm it was created as a triggered pipeline (the default mode).

help me find my pipeline in the Pipeline drop-down when creating a job task

Configure a scheduled trigger

Without a trigger, the job only runs when you manually start it. Adding a scheduled trigger automates pipeline refreshes so your Gold layer stays current.

  • In the Job details pane on the right side of the page, scroll down to the Schedules & Triggers section.
  • Click Add trigger.
  • Under Trigger type, choose Scheduled.
  • Keep the defaults of an active trigger that runs once per day.
  • Click Save.

Your trigger should now appear under Schedules & Triggers with an Active status indicator.

What are the other trigger types?

Lakeflow Jobs supports four trigger types: Scheduled (time-based), Table update (fires when source tables change), File arrival (fires when new files land in a Unity Catalog storage location), and Continuous (restarts after each completion). For a daily analytics refresh, Scheduled is the right fit.

Trigger status shows Paused instead of Active?

When creating or editing a trigger, you can toggle between Active and Paused in the Schedules & Triggers dialog. Make sure the Trigger Status is set to Active before clicking Save.

help me set my Lakeflow Job trigger to Active status

Run the job manually and validate end-to-end orchestration

The schedule handles future runs automatically. To validate the full orchestration right now without waiting until tomorrow, you can trigger a manual run.

  • Click Run now in the upper-right corner of the job page.
  • Click the Runs tab to monitor the execution.
  • Wait for the run to complete. Click the link in the Start time column to open the run details.
  • Click the pipeline task name to view task-level details including execution duration and status.

What should I check in the run details?

  • Task completion status: the task should show a green success indicator.
  • Execution duration: confirms how long the full Bronze-through-Gold refresh takes.
  • Pipeline event log: click into the pipeline run to see expectation results. All expectations should pass with zero violations for clean data.

You should see a successful run in the Runs tab with a green status. The pipeline task should complete without expectation violations.

Run failed or showing expectation violations?

If the run failed, click the task name in the run details to see the error. Common causes include the pipeline already running an update (only one update can run at a time) or a serverless compute quota being reached on Free Edition.

If you see expectation violations, the pipeline handled them as designed. Violations from expect_or_drop rules mean bad rows were dropped. Violations from expect_or_fail rules mean the pipeline stopped to protect downstream data. Check your source CSV files for null or invalid values.

help me debug why my Lakeflow Job run failed

Secret mission

Implement SCD Type-2 and Type-4 for Subscriber Dimensions

Your pipeline handles the current state of subscribers well, but what happens when a subscriber switches plans or migrates between segments? Production data platforms need to track these changes over time. In this secret mission, you'll implement two distinct Slowly Changing Dimension patterns to power historical reporting and migration analytics.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely on Databricks Free Edition at no cost. The pipeline and job use serverless compute within the daily free quota. If you exceed the daily quota, compute pauses until the next day, but your data and settings persist.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Unity Catalog schemas (bronze, silver, gold) with managed Delta tables and volumes.
  • One Lakeflow Spark Declarative Pipeline (telco pipeline with Bronze, Silver, and Gold layers).
  • One Lakeflow Job with a scheduled daily trigger.
  • One SQL Dashboard connected to the SQL Warehouse.

Keep everything running

No action needed. Choose this if you are still actively building or want to keep experimenting with your pipeline.

  • The scheduled job continues to trigger daily within your free quota.
  • If the job exceeds your daily compute quota, it waits until the next day to run. No data is lost.
  • All schemas, tables, notebooks, and the dashboard remain in your workspace.

Pause - I'll come back to this later

Stop the scheduled job from consuming your daily quota while preserving all configuration and data.

  • Click Jobs & Pipelines in the left sidebar.
  • Click your job name to open it.
  • In the job details panel, find Schedules and Triggers.
  • Click Pause to stop automatic runs while preserving the configuration.

Your pipeline, schemas, tables, notebooks, and dashboard all remain intact. Resume the schedule at any time by clicking the trigger and setting it back to active.

Delete - I don't want to use this again

Remove all project resources to start with a clean workspace.

Follow these steps in order to avoid dependency errors.

Delete the Lakeflow Job
  • Click Jobs & Pipelines in the left sidebar.
  • Click your job name to open the job page.
  • Click the kebab menu (three dots) next to the job name and select Delete job.
Delete the Lakeflow Spark Declarative Pipeline
  • Return to Jobs & Pipelines in the left sidebar.
  • Locate your pipeline in the list.
  • Click the kebab menu (three dots) next to the pipeline and select the delete option to remove it.
Delete the SQL Dashboard
  • Open your dashboard from the Dashboards section in the left sidebar.
  • Click the kebab menu (three dots) in the upper-right corner of the dashboard.
  • Click Move to trash.
Drop the Unity Catalog schemas
  • Open a SQL notebook or the SQL Editor and run these commands to drop all three schemas and their contents:
DROP SCHEMA IF EXISTS gold CASCADE;
DROP SCHEMA IF EXISTS silver CASCADE;
DROP SCHEMA IF EXISTS bronze CASCADE;

What does CASCADE do here?

The CASCADE option drops all tables, views, volumes, and functions inside each schema. This removes all managed Delta tables, the bronze.raw_data volume, and the uploaded CSV files in a single operation.

Delete the pipeline notebooks
  • In the Workspace browser, locate your pipeline notebooks (telco_pipeline.py, setup_schemas.sql, scd2_plan_history.py, scd4_segments_current.py, scd4_history.py).
  • Right-click each notebook and select Move to Trash.
  • To permanently remove them, open Trash and click Empty Trash.

Nice Work!

Nice Work!

Nice work! You just built a production-style governed telco lakehouse pipeline entirely on Databricks Free Edition.

You've learned how to:

  • Build a Lakeflow Spark Declarative Pipeline with data quality expectations that reject bad data at Bronze, Silver, and Gold layers using @dp.expect_or_fail and @dp.expect_all_or_drop.
  • Design a governed medallion architecture under Unity Catalog with end-to-end lineage, and visualize telco KPIs through a Databricks SQL dashboard connected to Gold-layer materialized views.
  • Orchestrate the entire pipeline with a scheduled Lakeflow Job that triggers a full Bronze-through-Gold refresh as a single managed task.
  • Secret Mission: Implemented SCD Type-2 for subscriber plan history using dp.create_auto_cdc_flow (time-travel reporting via CDC) and SCD Type-4 for segmentation (current dimension + migration history table for analytical lookback).

Ready to quiz yourself?