# Lakeflow pipelines on Databricks: Auto Loader, expectations and AUTO CDC

On Databricks, a pipeline is made in the Lakeflow Pipelines Editor with New, then ETL Pipeline, and its code is the same pyspark.pipelines code that runs on your computer. Lakeflow pipelines add what open-source Spark Declarative Pipelines lacks: expectations to warn, drop or fail on bad rows, and AUTO CDC to keep the newest state of each key from a stream of changes, even when changes arrive out of order. A triggered pipeline updates every table and stops, which suits a schedule.

- Page: https://tecyfy.com/data-engineering/course/medallion-pipelines/on-databricks
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 7: Pipelines: medallion with declarative pipelines, Lakeflow pipelines on Databricks
- Parts of this module: A first pipeline, bronze to gold (https://tecyfy.com/data-engineering/course/medallion-pipelines); Streaming tables, materialized views and full refresh (https://tecyfy.com/data-engineering/course/medallion-pipelines/refresh); Rules, quarantine and a quality gate (https://tecyfy.com/data-engineering/course/medallion-pipelines/expectations); Lakeflow pipelines on Databricks (https://tecyfy.com/data-engineering/course/medallion-pipelines/on-databricks); Build it: the shop's pipeline, checked (https://tecyfy.com/data-engineering/course/medallion-pipelines/build-it)

At the end of this part you can:

- Create a pipeline in the Lakeflow Pipelines Editor and run it
- Load files into bronze with Auto Loader, drop bad lines with expectations and keep 1 row per order with AUTO CDC
- Choose between triggered and continuous mode for a pipeline

[Parts 1 to 3](/data-engineering/course/medallion-pipelines) built the shop's pipeline on your computer: a bronze streaming table of every order line, a silver table of 1 row per order with a quarantine for bad lines, and gold spend per customer. This part builds the same pipeline on Databricks. It runs only there, so its code follows Databricks' docs. It uses the `landing` volume from [module 6](/data-engineering/course/getting-data-in/auto-loader), holding the 4 days of order files.

## Make the pipeline

Code written for the open-source `pyspark.pipelines` module runs on Databricks unchanged, so parts 1 to 3 would run once their paths pointed at a volume. Lakeflow pipelines add 2 things that make the shop's silver simpler: expectations and AUTO CDC. Neither is in Apache Spark.

First make a schema for the pipeline's tables. In a SQL editor, with your catalog selected, run `CREATE SCHEMA IF NOT EXISTS shop`. Then in the sidebar, click **New**, then **ETL Pipeline**. The Lakeflow Pipelines Editor opens. Name the pipeline `shop`, and beside the name choose your catalog and the `shop` schema: every table without a schema in its name goes there. Each layer below is its own file: **Add**, then **Transformation**.

## Bronze with Auto Loader

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

LANDING = "/Volumes/<catalog>/bronze/landing"

@dp.table(comment="Every line of every order file, once, as it arrived")
def bronze_orders():
    return (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("cloudFiles.schemaHints", "order_id BIGINT, amount DECIMAL(10,2), updated_at TIMESTAMP")
        .load(LANDING)
        .select("*", F.col("_metadata.file_name").alias("source_file"))
    )
```

Put your catalog's name in `LANDING`. There is no checkpoint location: the pipeline keeps a checkpoint for each flow itself, and Databricks' tutorial sets no schema location for Auto Loader in a pipeline either. As [module 6](/data-engineering/course/getting-data-in/auto-loader) showed, Auto Loader puts a value that does not fit, such as order 1003's `abc`, in `_rescued_data`.

## Silver with expectations and AUTO CDC

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

RULES = {
    "parsed": "_rescued_data IS NULL",
    "known_status": "status IN ('placed', 'paid', 'shipped', 'cancelled', 'refunded')",
    "positive_amount": "amount IS NOT NULL AND amount > 0",
}

dp.create_streaming_table(name="order_lines", expect_all_or_drop=RULES)

@dp.append_flow(target="order_lines")
def order_lines_flow():
    return spark.readStream.table("bronze_orders").select(
        "order_id", "customer_id", F.lower(F.trim("status")).alias("status"),
        "amount", "updated_at", "_rescued_data", "source_file",
    )

dp.create_streaming_table(name="orders", comment="1 row per order, in its newest state")
dp.create_auto_cdc_flow(
    target="orders",
    source="order_lines",
    keys=["order_id"],
    sequence_by=F.col("updated_at"),
    except_column_list=["_rescued_data", "source_file"],
)
```

`order_lines` is a streaming table whose rules drop every line that breaks one, so it holds good lines only. Its rules are part 3's, with `refunded` now a known status, and with the NULL case written into `positive_amount`, since Databricks says a rule must come out true or false for every row. Each rule's count of dropped lines shows in the pipeline's **Data quality** tab. To keep the bad lines instead, build part 3's quarantine, which is Databricks' own pattern, with `@dp.expect_all` to count them.

`orders` is filled by an AUTO CDC flow. For each `order_id` it keeps 1 row, the line with the latest `updated_at`, and updates the row as new lines arrive. That is part 1's `max_by`, with 2 differences. It applies only the lines that are new since the last update, rather than recomputing every order. And a line older than the one it would replace is dropped, which matters because Auto Loader does not promise to load files in order. `except_column_list` leaves the 2 bookkeeping columns out of `orders`. Its default, SCD type 1, keeps no history; `stored_as_scd_type="2"` keeps every version, as [a slowly changing dimension](https://tecyfy.com/concepts/slowly-changing-dimensions) does.

## Gold, and the first run

Add a SQL file with part 1's `customer_spend`, changing its filter to `WHERE status NOT IN ('cancelled', 'refunded')` now that refunds are good lines. Click **Dry run** to check the code without updating any table, then **Run pipeline**. The pipeline graph beside the code shows the 4 tables in the order they ran, and the panel below shows each table's rows. Run it again and nothing new is read.

Every table here is a Delta table. On serverless compute, the only compute Free Edition has, a materialized view such as `customer_spend` is refreshed incrementally when Databricks finds that cheaper than a full recompute. The editor's table list says which it did. **Run pipeline with full table refresh** is part 2's `--full-refresh-all`.

## Triggered or continuous

A pipeline runs in 1 of 2 modes:

- **Triggered**, the default, processes the data that is there when the update starts, then stops. It costs only while it runs, and suits updates every 10 minutes, hourly or daily. The shop's daily files need nothing more.
- **Continuous** keeps compute running and processes new data as it arrives, for tables that must be fresh within seconds or minutes. For a continuous pipeline, Databricks recommends running it from a continuous job rather than setting the pipeline's own mode.

Free Edition allows 1 active pipeline per pipeline type. Module 10 runs this pipeline on a schedule, as a task of a job.

## Check yourself

1. Auto Loader does not promise the order in which it loads files. The file that ships order 1004 is loaded before the file that placed it. What does the AUTO CDC flow keep for 1004?

   A. The placed line, because it was loaded last
   B. Both lines, as 2 rows
   C. The shipped line, because sequence_by orders the changes by updated_at, not by when they arrived
   D. The flow fails on the late line

### Answers

1. C. AUTO CDC uses the sequence_by column to put changes in order, and with SCD type 1 a change older than the row it would replace is dropped. Silver in part 1 got the same result with max_by, but recomputed every order on each run; AUTO CDC applies only the new changes.

## Questions

### What is the difference between Spark Declarative Pipelines and Lakeflow pipelines?

Spark Declarative Pipelines is the open-source framework in Apache Spark, from Spark 4.1. Lakeflow pipelines are Databricks' product built on it, and code written for the open-source pyspark.pipelines module runs on Databricks without change. Databricks adds expectations, the AUTO CDC APIs, the Lakeflow Pipelines Editor, serverless compute with incremental refresh of materialized views, and scheduling with jobs. Its old name was Delta Live Tables, with the dlt module.

### Should a Databricks pipeline run in triggered or continuous mode?

Triggered, unless data must be fresh within seconds or minutes. A triggered update processes the data available when it starts, then stops, so it costs only while it runs and suits updates every 10 minutes, hourly or daily. Continuous mode keeps compute running and processes data as it arrives. For a continuous pipeline, Databricks recommends a continuous job rather than the pipeline's own continuous setting.

## Sources

- [Tutorial: Build an ETL pipeline using change data capture](https://docs.databricks.com/aws/en/ldp/tutorial-pipelines)
- [Lakeflow pipelines Python language reference](https://docs.databricks.com/aws/en/ldp/developer/python-ref)
- [The AUTO CDC APIs: simplify change data capture with pipelines](https://docs.databricks.com/aws/en/ldp/cdc)
- [create_auto_cdc_flow](https://docs.databricks.com/aws/en/ldp/developer/ldp-python-ref-apply-changes)
- [Triggered vs. continuous pipeline mode](https://docs.databricks.com/aws/en/ldp/pipeline-mode)
- [Incremental refresh for materialized views](https://docs.databricks.com/aws/en/optimizations/incremental-refresh)
- [Run an update in a pipeline](https://docs.databricks.com/aws/en/ldp/updates)
- [Develop and debug ETL pipelines with the Lakeflow Pipelines Editor](https://docs.databricks.com/aws/en/ldp/multi-file-editor)
- [Databricks Free Edition limitations](https://docs.databricks.com/aws/en/getting-started/free-edition-limitations)