Medallion ETL Pipeline with Databricks DAB

Introduction

30 Second Summary

Every time you place an online order, that transaction generates raw data. Turning that raw data into something useful for business decisions requires a pipeline that cleans, validates, and organizes it automatically.

In this project, you will build a production-ready ETL pipeline using Lakeflow Spark Declarative Pipelines on Azure Databricks. You'll transform raw retail sales data through three layers of refinement and deploy the entire thing as code using Databricks Asset Bundles.

What You'll Build

A 3-layer medallion pipeline that ingests raw JSON sales orders, validates them with data quality expectations, and aggregates them into analytics-ready gold tables.

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

• A Lakeflow SDP pipeline with streaming tables and materialized views across bronze, silver, and gold layers.
• Data quality expectations that automatically drop invalid rows and emit tracking metrics.
• A DAB bundle project with deployment targets, local pytest tests, and a repeatable deploy workflow.
• Secret Mission: Add a second data source, build a cross-table join, and validate referential integrity with expectations.

Are there any prerequisites?

This project assumes you already have an Azure account with free credits and a Premium-tier Databricks workspace provisioned. You also need macOS with Homebrew installed.

If you haven't set these up yet, follow Azure's documentation to create your free account and provision a Databricks workspace before starting Step 1.

Before We Start

Before We Start...

Install the Databricks CLI and Authenticate

Before you can build and deploy a pipeline, you need a way to talk to your Databricks workspace from the terminal.

The Databricks CLI gives you that bridge. In this step, you'll install it, authenticate with your workspace using OAuth, and install uv (the Python package runner you'll use to run tests later).

In this step, get ready to:

• Install the Databricks CLI and verify it's version 0.283.0 or above.
• Authenticate the CLI with your Azure Databricks workspace using OAuth U2M.
• Install uv, the Python package runner required for local tests.

Install the Databricks CLI via Homebrew

The Databricks CLI is a command-line tool that lets you manage pipelines, bundles, and workspace resources directly from your terminal.

• Press Cmd+Space to open Spotlight Search.
• Type Terminal and press Enter to open it.

• Add the Databricks Homebrew tap and install the CLI by running these commands:

brew tap databricks/tap
brew trust databricks/tap
brew install databricks

What do these commands do?

• brew tap databricks/tap registers the Databricks formula repository with Homebrew so it knows where to find the package.
• brew trust databricks/tap marks the tap as trusted so Homebrew allows installation without additional prompts.
• brew install databricks downloads and installs the Databricks CLI binary.

• Verify the installation by running:

databricks -v

You should see a version number of 0.205.0 or higher. The project requires version 0.283.0+ for the pipeline scaffolding command you'll use in the next step.

✔️ I see version 0.283.0 or higher

You're all set. The Databricks CLI is ready to go.

ⓧ I see a version lower than 0.283.0

You need to upgrade to get the databricks pipelines init command.

• Upgrade the CLI by running:

brew upgrade databricks

• Run databricks -v again to confirm the version is 0.283.0 or above.

ⓧ Command not found

Homebrew may not have linked the binary to your PATH.

• Close your terminal and open a new one.
• Run databricks -v again.
• If it still fails, make sure Homebrew's bin directory is on your PATH. Check the output of brew --prefix and ensure that path plus /bin is in your shell's PATH.

Configure OAuth U2M authentication

OAuth U2M (user-to-machine) authentication lets the CLI manage short-lived tokens on your behalf. Tokens expire in under an hour, reducing risk if one is accidentally exposed.

• Find your Azure Databricks workspace URL. Open your workspace in a browser and copy the URL from the address bar. It looks like https://adb-1234567890123456.7.azuredatabricks.net.
• Run the login command, replacing the placeholder with your actual workspace URL:

databricks auth login --host [[WORKSPACE_URL="https://adb-1234567890123456.7.azuredatabricks.net"]]

What happens next?

A browser window opens and asks you to sign in to your Azure Databricks workspace. After you sign in, the CLI caches an OAuth token in your local token cache.

The CLI also prompts you to save a configuration profile name. Press Enter to accept the suggested profile name.

• Complete the browser sign-in flow when the window opens.
• Press Enter to accept the suggested profile name when prompted.

• Verify your authentication is working by running:

databricks auth token --host [[WORKSPACE_URL="https://adb-1234567890123456.7.azuredatabricks.net"]]

You should see a JSON response containing an access_token field and an expiration timestamp. This confirms the CLI can reach your workspace.

Browser didn't open or login failed?

• Make sure the workspace URL starts with https:// and ends with .azuredatabricks.net. A trailing slash can sometimes cause issues.
• Confirm your Azure account has access to the workspace. If you see a permissions error, check with your workspace admin.
• If you're on a corporate network with a proxy, try from a personal network connection.

Help me troubleshoot the Databricks CLI OAuth login.

Install uv

The pipeline bundle project uses uv to manage Python dependencies and run tests locally. It's a fast Python package runner that handles virtual environments automatically.

• Install uv via Homebrew by running:

brew install uv

• Verify it installed correctly by running:

uv --version

You should see a version number printed in the terminal.

✔️ I see a version number

Nice. uv is installed and ready for running tests later.

ⓧ Command not found

Homebrew may not have linked the binary to your PATH.

• Close your terminal and open a new one.
• Run uv --version again.
• If it still doesn't work, run brew link uv to ensure the binary is linked to your PATH.

Help me troubleshoot my uv installation on macOS.

Your tooling is ready. Next up, you'll use the Databricks CLI to scaffold a complete pipeline bundle project with all the files and configuration you need.

Initialize the Pipeline Bundle Project

Your CLI is authenticated and ready to talk to your workspace. Now you need a project structure that defines your pipeline as code.

Databricks provides a scaffold command that generates a complete bundle project with pipeline definitions, deployment targets, and test infrastructure. This is the foundation you will build your medallion pipeline on top of.

In this step, get ready to:

• Scaffold a pipeline bundle project using the Databricks CLI.
• Explore the generated project structure and understand what each directory does.
• Validate the bundle configuration to confirm everything is wired up correctly.

Scaffold the pipeline bundle project

The databricks pipelines init command generates a ready-to-deploy bundle project specifically designed for Lakeflow Spark Declarative Pipelines. It creates the configuration files, resource definitions, and source directories you need.

Why pipelines init instead of bundle init?

The databricks pipelines init command scaffolds a bundle specifically designed for pipeline development. It includes pipeline resource definitions, pytest setup, and uv configuration out of the box.

The generic databricks bundle init command creates a job-focused bundle without pipeline resources. Since you are building a pipeline, the pipeline-specific template saves you from manual setup.

• Switch to your Desktop in the terminal by running this command:

cd ~/Desktop

• Scaffold your pipeline bundle project by running:

databricks pipelines init

What will this command ask me?

The pipelines init command launches an interactive wizard with several prompts. Answer each one as described in the bullets below.

• For Unique name for this project, type medallion_pipeline and press Enter.
• For Initial catalog, type the name of an existing Unity Catalog catalog in your workspace (for example, main or whichever catalog you created) and press Enter.
• For Use a personal schema for each user working on this project?, select yes.
• For Initial language for this project, select python.

The CLI creates a medallion_pipeline folder on your Desktop containing the full project scaffold.

• Move into your new project folder by running:

cd medallion_pipeline

• Confirm the project files exist by running:

ls

You should see databricks.yml, pyproject.toml, README.md, resources/, src/, and tests/ listed in the output.

Command not found or version error?

The databricks pipelines init command requires Databricks CLI version 0.283.0 or above. Run databricks -v to check your version.

If your version is lower, update the CLI with brew upgrade databricks and try again.

Help me troubleshoot this error

Explore the generated project structure

The template generates a complete Declarative Automation Bundles project with everything you need to define, test, and deploy a pipeline.

• View the full project tree by running:

find . -type f | head -20

What did the template create?

• databricks.yml is the bundle configuration file. It defines the bundle name, catalog and schema variables, and deployment targets (dev and prod).
• resources/ contains pipeline and job resource definitions as YAML files. These tell Databricks what resources to create when you deploy.
• src/ holds the pipeline source code. This is where your Python transformation logic lives.
• tests/ is the directory for local pytest tests that validate your pipeline logic without needing a running cluster.
• pyproject.toml configures Python dependencies and test settings for local testing via uv.

What are deployment targets?

The databricks.yml file defines a dev target and a prod target. The dev target uses mode: development, which prefixes all resources with your username and pauses any scheduled jobs automatically.

This means you can deploy freely during development without affecting production resources or accidentally running scheduled jobs.

Validate the bundle configuration

Before deploying anything, you should verify that your bundle configuration is valid. The validate command checks that your YAML is correct and that all referenced files exist.

• Validate your bundle by running:

databricks bundle validate

You should see a summary of the bundle configuration printed to your terminal with no errors.

What does this command check?

The databricks bundle validate command parses your databricks.yml and all included resource files. It checks for YAML syntax errors, missing file references, and invalid configuration values.

If validation succeeds, you will see a JSON summary of the bundle including its name, targets, and resources. This confirms everything is wired up correctly before you deploy.

Seeing validation errors?

Make sure you are running the command from inside the medallion_pipeline folder (the directory containing databricks.yml).

If you see an authentication error, your OAuth token may have expired. Re-run the auth login command from Step 1 to refresh it.

Help me fix this validation error

Your bundle project is scaffolded and validated. Next up, you will replace the sample pipeline code with your own medallion architecture transformations.

Build the Medallion Pipeline (Bronze and Silver)

Your bundle project is scaffolded and validated. Now it's time to write the actual pipeline code that transforms raw data into clean, structured tables.

The medallion architecture organizes data into layers: bronze (raw ingestion), silver (cleaned and validated), and gold (aggregated for analytics). In this step, you'll build the first two layers using Lakeflow Spark Declarative Pipelines.

In this step, get ready to:

• Create a bronze streaming table that ingests raw JSON data using Auto Loader.
• Create a silver streaming table with data quality expectations that drop invalid rows.
• Review how the pipeline decorators define table dependencies.

Create the bronze streaming table

The bronze layer is responsible for ingesting raw data exactly as it arrives. Auto Loader handles this by monitoring a file path and automatically picking up new JSON files as they land.

• Open src/pipeline.py in your editor.
• Replace the entire placeholder content with the following code:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, date_format

# Define the bronze layer: raw ingestion from JSON files
@dp.table()
def bronze_orders():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("/databricks-datasets/retail-org/sales_orders")
    )

What does this code do?

The from pyspark import pipelines as dp import gives you access to the Lakeflow SDP module. This module is only available inside the pipeline execution context on Databricks, not in standalone Python.

The @dp.table() decorator tells Lakeflow to create a streaming table from whatever the function returns. The function uses Auto Loader (cloudFiles format) to incrementally read new JSON files from the built-in retail dataset at /databricks-datasets/retail-org/sales_orders.

• Save src/pipeline.py.
• You won't see a change yet. The next substep adds the silver layer, and a later step deploys the full pipeline to Databricks.

Why Auto Loader instead of a static read?

Auto Loader uses the cloudFiles source to automatically detect and process new files as they arrive in storage. This means your bronze table stays up to date without you writing any scheduling or file-tracking logic.

Inside Lakeflow SDP, the pipeline manages schema inference and checkpointing automatically. You just point it at the path and let the framework handle the rest.

Create the silver streaming table with data quality expectations

The silver layer cleans and validates data coming from bronze. You'll add a data quality expectation that automatically drops any row where the order_datetime field is null or empty.

• In src/pipeline.py, add the following code below the bronze_orders function:

# Define the silver layer: clean and validate incoming orders
@dp.table()
@dp.expect_or_drop("valid_date", "order_datetime IS NOT NULL AND length(order_datetime) > 0")
def silver_orders():
    return (
        spark.readStream
        .table("bronze_orders")
        .select(
            "customer_id",
            "order_number",
            col("order_datetime").cast("int").cast("timestamp").alias("order_timestamp"),
            "quantity",
            col("order_datetime").cast("int").cast("timestamp").cast("date").alias("order_date")
        )
    )

What does this code do?

The @dp.expect_or_drop() decorator defines a data quality expectation. Any row that fails the condition order_datetime IS NOT NULL AND length(order_datetime) > 0 is silently dropped from the output table.

The function reads from bronze_orders as a stream, creating an automatic dependency between the two tables. It selects only the columns needed downstream and casts the raw Unix timestamp string into a proper timestamp and date type for analytics.

• Save src/pipeline.py.
• You won't see runtime output yet. The pipeline runs on Databricks serverless compute, which you'll deploy in the next step.

Seeing a syntax error in your editor?

Your editor may underline spark as undefined. This is expected. The spark variable and the pyspark.pipelines module are only available inside the Databricks pipeline execution context, not in standalone Python.

Make sure your @dp.table() and @dp.expect_or_drop() decorators are on separate lines directly above def silver_orders(): with no blank lines between them.

If you need more help, ask about pipeline syntax errors.

✔️ Awesome, I've got everything!

Great. Double check you've saved src/pipeline.py with both the bronze_orders and silver_orders functions.

ⓧ I'd like to double check the full code

Here is the complete src/pipeline.py file as it should look right now:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, date_format

@dp.table()
def bronze_orders():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("/databricks-datasets/retail-org/sales_orders")
    )

@dp.table()
@dp.expect_or_drop("valid_date", "order_datetime IS NOT NULL AND length(order_datetime) > 0")
def silver_orders():
    return (
        spark.readStream
        .table("bronze_orders")
        .select(
            "customer_id",
            "order_number",
            col("order_datetime").cast("int").cast("timestamp").alias("order_timestamp"),
            "quantity",
            col("order_datetime").cast("int").cast("timestamp").cast("date").alias("order_date")
        )
    )

Review how the decorators work together

Your pipeline now has two tables with an automatic dependency chain. Lakeflow SDP uses the decorator pattern to turn plain Python functions into managed pipeline assets.

How do the decorators define the pipeline?

• @dp.table() tells Lakeflow to create a streaming table from the function's return value. The function name becomes the table name.
• @dp.expect_or_drop("valid_date", "...") attaches a data quality rule. Rows that fail the SQL condition are dropped before they reach the output table. Lakeflow logs how many rows passed and failed in the pipeline event log.
• spark.readStream.table("bronze_orders") in the silver function creates a dependency on the bronze table. Lakeflow uses this to determine the correct execution order automatically.

You did not write any scheduling logic, checkpoint management, or orchestration code. The framework infers the full DAG from the function references and handles execution for you.

Your bronze and silver layers are defined. Next up, you'll add the gold aggregation layer and deploy the entire pipeline to your Databricks workspace.

Build the Gold Layer and Deploy

Your bronze and silver layers handle raw ingestion and data cleaning. Now you need the final piece: a gold layer that turns cleaned data into business-ready analytics.

In this step, you will create a materialized view that aggregates daily order counts by state, deploy the entire pipeline bundle to your Azure Databricks workspace, and run it end-to-end.

In this step, get ready to:

• Create a gold materialized view that aggregates daily order counts by state.
• Deploy the complete pipeline bundle to your workspace.
• Run the pipeline and verify all 3 medallion layers populate in the Databricks UI.

Create the gold materialized view

The gold layer in a medallion architecture is where cleaned data gets shaped into answers. Your gold table will join orders with customer data and aggregate daily order counts grouped by state.

• In src/pipeline.py, scroll to the bottom of the file (below the silver_orders function).
• Add the following gold materialized view below the silver_orders function:

@dp.materialized_view()
def gold_daily_orders_by_state():
    return (
        spark.read.table("silver_orders")
        .join(
            spark.read.format("csv")
            .option("header", True)
            .load("/databricks-datasets/retail-org/customers"),
            "customer_id"
        )
        .groupBy("state", "order_date")
        .count()
        .withColumnRenamed("count", "order_count")
    )

What does this code do?

This function uses the @dp.materialized_view() decorator instead of @dp.table(). Materialized views are the recommended pattern for gold-layer aggregations because they recompute from scratch each time rather than processing incrementally.

• spark.read.table("silver_orders") reads the cleaned silver table as a batch source.
• The .join() brings in customer data from /databricks-datasets/retail-org/customers (a CSV file with a header row) and matches on customer_id.
• .groupBy("state", "order_date") groups records by customer state and order date, then .count() counts orders per group.
• .withColumnRenamed("count", "order_count") gives the count column a descriptive name for downstream consumers.

• Save src/pipeline.py.

You won't see a change yet. The next substep deploys and runs the full pipeline so you can verify all three layers together.

Seeing a syntax error?

• Make sure the gold_daily_orders_by_state function is defined below silver_orders with no indentation errors.
• Check that the .join() has a closing parenthesis matching the spark.read.format("csv") block.
• Confirm you used @dp.materialized_view() (with parentheses), not @dp.materialized_view.

Help me debug a syntax error in my gold_daily_orders_by_state materialized view.

✔️ Awesome, I've got everything!

Great. Double check you have saved src/pipeline.py before moving on to deployment.

ⓧ I'd like to double check the full code

from pyspark import pipelines as dp
from pyspark.sql.functions import col, date_format

@dp.table()
def bronze_orders():
    return (
        spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .load("/databricks-datasets/retail-org/sales_orders")
    )

@dp.table()
@dp.expect_or_drop("valid_date", "order_datetime IS NOT NULL AND length(order_datetime) > 0")
def silver_orders():
    return (
        spark.readStream
        .table("bronze_orders")
        .select(
            "customer_id",
            "order_number",
            col("order_datetime").cast("int").cast("timestamp").alias("order_timestamp"),
            "quantity",
            col("order_datetime").cast("int").cast("timestamp").cast("date").alias("order_date")
        )
    )

@dp.materialized_view()
def gold_daily_orders_by_state():
    return (
        spark.read.table("silver_orders")
        .join(
            spark.read.format("csv")
            .option("header", True)
            .load("/databricks-datasets/retail-org/customers"),
            "customer_id"
        )
        .groupBy("state", "order_date")
        .count()
        .withColumnRenamed("count", "order_count")
    )

Deploy the pipeline bundle to your workspace

With all three layers defined in code, it is time to push the entire bundle to your Azure Databricks workspace. The deploy command uploads your source files and creates the pipeline resource.

• Deploy the complete pipeline bundle to your dev target by running this command from the project root:

databricks bundle deploy --target dev

What does this command do?

This command packages your databricks.yml configuration, your src/ source files, and your resource definitions, then pushes them to the workspace under the dev target. It creates or updates the Lakeflow SDP pipeline resource in your workspace.

• Wait for the command to finish. You should see a summary showing your bundle name, target, workspace host, user, and a validation confirmation.
• Confirm the deployment by opening your Azure Databricks workspace sidebar and clicking Workspace.
• Navigate into Users > your username > .bundle folder. You should see your bundle project listed there.

Deploy command failed?

• Run databricks bundle validate first to check for configuration errors.
• Confirm your CLI is still authenticated. If your OAuth token expired, re-authenticate with the CLI.
• Make sure you are running the command from the project root directory (where databricks.yml lives).

Help me troubleshoot a failed databricks bundle deploy command.

Run the pipeline and verify all 3 layers

The bundle is deployed but the pipeline has not processed any data yet. Running the pipeline triggers Auto Loader to ingest, transformations to apply, and the gold view to materialize.

• Trigger the pipeline run by running this command:

databricks pipelines run

Why no pipeline name?

Since there is only one pipeline defined in your project, the CLI auto-selects it. You do not need to specify a pipeline name.

• Wait for the run to complete. The terminal displays progress updates and a final status.
• Look for the Update URL value that appears in your terminal output.
• Copy the Update URL and paste it into your web browser to open the pipeline run details in your Azure Databricks workspace.
• In the pipeline run view, confirm that all three tables appear: bronze_orders, silver_orders, and gold_daily_orders_by_state.
• Click on gold_daily_orders_by_state to verify it shows row counts and the columns state, order_date, and order_count.

What should I see?

The pipeline graph in the workspace UI shows three nodes connected by arrows: bronze_orders feeds into silver_orders, which feeds into gold_daily_orders_by_state. Each node should show a green checkmark indicating successful completion.

The gold table joins your cleaned orders with the customers CSV to pull in the state field, then groups by state and date. You should see aggregated rows with counts per state per day.

Pipeline run failed or tables are missing?

• If only bronze and silver appear but gold is missing, check that src/pipeline.py was saved before you ran the deploy command. Re-deploy and re-run if needed.
• If you see an error about the Advanced product edition, confirm your pipeline configuration uses the Advanced edition (required for expectations).
• If the run times out or shows no progress, verify your workspace has serverless compute enabled.

Help me debug my pipeline run that is not completing successfully.

Your complete medallion pipeline is now running in the cloud. All three layers process data from raw ingestion through to business-ready aggregations. Next up, you will write local tests to validate the transformation logic and check data quality metrics.

Run Local Tests and Verify Data Quality

Your pipeline is deployed and running. But how do you know the transformation logic is correct before pushing changes to production?

In this step, you will write local pytest tests that validate your pipeline's transformation logic without needing a Spark cluster. You will also query the pipeline event log directly in your workspace to confirm that data quality expectations are enforcing the rules you set.

In this step, get ready to:

• Write pytest tests that validate date parsing, null filtering, and aggregation column logic.
• Run the tests locally with uv run pytest.
• Query the pipeline event log to verify expectation metrics.

Write pytest tests for your pipeline logic

The pyspark.pipelines module only runs inside the Databricks pipeline execution context. You cannot import dp in a local Python environment. Instead, local tests validate the pure transformation logic (date parsing, null filtering conditions) using standard Python.

• Open tests/test_pipeline.py in your editor (this file already exists in the scaffolded tests/ directory).
• Replace any existing content with the following test code:

import pytest
from datetime import date


def test_order_date_parsing():
    """Test that unix timestamp strings are correctly parsed to dates."""
    import datetime
    # Simulate the transformation logic: cast string -> int -> timestamp -> date
    sample_unix = 1672531200  # 2023-01-01 00:00:00 UTC
    expected_date = datetime.date(2023, 1, 1)
    from datetime import datetime as dt, timezone
    result = dt.fromtimestamp(sample_unix, tz=timezone.utc).date()
    assert result == expected_date


def test_null_filter_logic():
    """Test that null or empty order_datetime values would be filtered."""
    invalid_values = [None, "", "   "]
    for val in invalid_values:
        # The expectation "order_datetime IS NOT NULL AND length(order_datetime) > 0"
        # should drop rows where order_datetime is None or empty
        if val is None:
            assert True  # NULL fails IS NOT NULL
        else:
            assert len(val.strip()) == 0  # Empty/whitespace fails length > 0


def test_aggregation_columns():
    """Test that the gold table would have the expected column names."""
    expected_columns = {"state", "order_date", "order_count"}
    # This validates our withColumnRenamed produces the right schema
    assert "order_count" in expected_columns
    assert "state" in expected_columns

What does this code do?

These three tests mirror the transformation logic in your pipeline without importing any Databricks-specific modules:

• test_order_date_parsing simulates the cast("int").cast("timestamp").cast("date") chain from silver_orders. It proves that a known Unix timestamp correctly converts to the expected calendar date.
• test_null_filter_logic validates that your expect_or_drop expression would correctly identify null, empty, and whitespace-only values as invalid.
• test_aggregation_columns confirms that your gold table schema uses the correct column names after the withColumnRenamed call.

Why test locally without Spark?

Databricks recommends testing transformation logic (date parsing, filtering conditions, schema expectations) in pure Python. This catches bugs before you deploy and run a pipeline, saving both time and compute costs.

The pyspark.pipelines module is only available within the pipeline execution context on Databricks. It is not importable in standalone Python. Local tests focus on the logic, not the framework.

• Save tests/test_pipeline.py.

Run the tests locally

• Run your test suite from the project root by running this command:

uv run pytest

You should see output showing all 3 tests passing, similar to:

========================= 3 passed in 0.05s =========================

Tests failing or command not found?

• If you see a module import error, make sure you are running the command from your bundle project root (the folder containing databricks.yml).
• If uv is not found, revisit Step 1 where you installed it.
• If a test assertion fails, double-check that you copied the test file exactly as shown above.

Still stuck? Help me debug my failing pytest tests for the Databricks pipeline.

Query the pipeline event log for expectation metrics

Your silver_orders table uses @dp.expect_or_drop to discard rows with null or empty order_datetime values. Every pipeline run records exactly how many rows passed and failed that expectation in the event log. Now you will query those metrics.

• In your Azure Databricks workspace, click SQL Editor in the left sidebar.
• Paste the following SQL query into the editor:

CREATE OR REPLACE TEMP VIEW latest_update AS
SELECT origin.update_id AS id
FROM event_log(TABLE(silver_orders))
WHERE event_type = 'create_update'
ORDER BY timestamp DESC
LIMIT 1;

WITH expectations_parsed AS (
  SELECT
    explode(
      from_json(
        details:flow_progress.data_quality.expectations,
        "array<struct<name: string, dataset: string, passed_records: int, failed_records: int>>"
      )
    ) row_expectations
  FROM
    event_log(TABLE(silver_orders)),
    latest_update
  WHERE
    event_type = 'flow_progress'
    AND origin.update_id = latest_update.id
)
SELECT
  row_expectations.dataset as dataset,
  row_expectations.name as expectation,
  SUM(row_expectations.passed_records) as passing_records,
  SUM(row_expectations.failed_records) as failing_records
FROM expectations_parsed
GROUP BY
  row_expectations.dataset,
  row_expectations.name;

• Click Run (or press Ctrl+Enter).

What does this query do?

The event_log() table-valued function returns all pipeline events for a given table. The query first identifies the most recent pipeline update, then extracts the data quality expectations array from the flow_progress events.

The result shows your valid_date expectation with two columns: passing_records (rows that had a valid order_datetime) and failing_records (rows that were dropped because they failed the check). This confirms your data quality gate is actively filtering bad data.

You should see a result row showing the valid_date expectation for the silver_orders dataset, with a count of passing records and a count of failing (dropped) records.

Query returns no rows?

• Make sure you ran the pipeline at least once in Step 4. The event log only contains metrics after a successful pipeline run.
• Confirm the table name silver_orders matches what your pipeline created. If you chose a specific catalog and schema during bundle setup, qualify the table name (e.g., catalog.schema.silver_orders).
• The event_log() function can only be called by the owner of the table. Make sure you are logged in as the same user who deployed the pipeline.

Still stuck? Help me query the pipeline event log for my data quality expectation metrics.

✔️ Awesome, I've got everything!

Great. Make sure you have saved tests/test_pipeline.py and confirmed all 3 tests pass locally.

ⓧ I'd like to double check the full code

Here is the complete tests/test_pipeline.py file:

import pytest
from datetime import date


def test_order_date_parsing():
    """Test that unix timestamp strings are correctly parsed to dates."""
    import datetime
    # Simulate the transformation logic: cast string -> int -> timestamp -> date
    sample_unix = 1672531200  # 2023-01-01 00:00:00 UTC
    expected_date = datetime.date(2023, 1, 1)
    from datetime import datetime as dt, timezone
    result = dt.fromtimestamp(sample_unix, tz=timezone.utc).date()
    assert result == expected_date


def test_null_filter_logic():
    """Test that null or empty order_datetime values would be filtered."""
    invalid_values = [None, "", "   "]
    for val in invalid_values:
        # The expectation "order_datetime IS NOT NULL AND length(order_datetime) > 0"
        # should drop rows where order_datetime is None or empty
        if val is None:
            assert True  # NULL fails IS NOT NULL
        else:
            assert len(val.strip()) == 0  # Empty/whitespace fails length > 0


def test_aggregation_columns():
    """Test that the gold table would have the expected column names."""
    expected_columns = {"state", "order_date", "order_count"}
    # This validates our withColumnRenamed produces the right schema
    assert "order_count" in expected_columns
    assert "state" in expected_columns

You now have a full testing workflow: local pytest tests validate your transformation logic before deployment, and the event log confirms your data quality expectations are working in production. Next up, the Secret Mission will extend your pipeline with a second data source and a cross-table join.

Secret mission

Add Customers and a Join Table

Your pipeline currently processes orders in isolation. Real analytics needs customer context. In this mission, you'll add a second data source, join it with your orders, and validate referential integrity with expectations.

Clean Up Your Resources

Clean Up Your Resources

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

This project uses serverless compute on Azure Databricks, which consumes DBUs charged against your Azure free credits.

Cost warning

Serverless pipeline runs consume DBUs from your Azure free credits. A single run of this pipeline costs approximately $0.10-0.50 in DBUs.

The dev deployment mode pauses the daily job schedule by default, so no ongoing costs accrue unless you manually trigger additional runs. If you want to avoid any risk of accidental charges, choose Pause or Delete below.

Resources you used:

• One Lakeflow SDP pipeline (serverless) deployed to your Azure Databricks workspace.
• Streaming tables and materialized views in Unity Catalog (your specified catalog and schema).
• Bundle deployment files stored in your workspace.
• Local bundle project files on your Desktop.
• Databricks CLI installed via Homebrew.
• uv binaries installed on your machine.

Keep everything running

No action needed. Choose this if you are still actively experimenting with your pipeline or want to add more tables.

• Your pipeline remains deployed in its current state.
• The dev deployment mode pauses the daily job schedule by default, so no ongoing DBU costs accrue unless you manually trigger runs.
• The Databricks CLI, uv, and local project files stay on your machine for future use.

Pause - I'll come back to this later

The dev target already pauses the job schedule, so your pipeline will not run automatically. No additional action is needed to pause costs.

• Your deployed pipeline, tables, and workspace files remain intact.
• No DBUs are consumed unless you manually trigger a pipeline run.
• When you are ready to resume, navigate to your project folder and run your pipeline again.

Delete - I don't want to use this again

Remove all deployed resources and local project files to start completely fresh.

• Destroy all deployed bundle resources (pipeline, jobs, and workspace files) by running this command from your project folder:

databricks bundle destroy --target dev

What does this command do?

This removes the Lakeflow SDP pipeline, all associated streaming tables and materialized views, any scheduled jobs, and the workspace deployment files. Your Unity Catalog schema itself is not deleted, but the tables created by this pipeline are removed.

• Delete the local project folder by running:

cd ~/Desktop
rm -rf [[PROJECT_FOLDER="your-project-folder-name"]]

• Uninstall the Databricks CLI by running:

brew uninstall databricks

• Remove uv from your machine by running:

rm ~/.local/bin/uv ~/.local/bin/uvx

That's a Wrap!

That's a Wrap!

Nice work! You just built and deployed a production-ready medallion ETL pipeline on Azure Databricks, managed entirely through code with Databricks Asset Bundles.

You've learned how to:

• Install and authenticate the Databricks CLI, then scaffold a production-style bundle project with databricks pipelines init.
• Build a 3-layer medallion pipeline (bronze, silver, gold) using Lakeflow Spark Declarative Pipelines with data quality expectations that drop invalid rows and emit metrics.
• Deploy and run the pipeline via Databricks Asset Bundles and validate transformation logic with local pytest tests.
• Secret Mission: Added a second data source, built a cross-table join, and validated referential integrity with expectations.

Ready to quiz yourself?