Delta Live Tables (DLT) is the former name of what Databricks now calls Lakeflow Declarative Pipelines.

In other words: Lakeflow Declarative Pipelines is the current evolution and product name of DLT.

Lakeflow Declarative Pipelines is a data-pipeline framework in Databricks for defining, executing, and maintaining data processing pipelines using a declarative programming model.

The key idea is:
You define the desired state and data relationships of your datasets, rather than explicitly controlling every execution step.
你描述“数据应该是什么样、数据之间是什么关系、数据质量应该满足什么条件”，而不是手工控制每一个执行步骤。

A Declarative Pipeline can contain:

• data sources
• data flows
• streaming tables
• materialized views
• transformations
• data-quality rules
• dependencies between datasets
• pipeline configuration
• operational and monitoring information

The pipeline engine uses the definitions to construct a dependency graph / DAG and determine how the pipeline should be executed.

In current Databricks terminology, the technology formerly known as Delta Live Tables (DLT) has evolved into Lakeflow Declarative Pipelines. The underlying open framework is also referred to as Apache Spark Declarative Pipelines (SDP).

Our Data platform:

ADLS
↓
Bronze
↓
Silver
↓
Gold

Traditional data engineering flow looks this:

ADF Trigger
      ↓
Databricks Job
      ↓
Notebook 1
      ↓
Notebook 2
      ↓
Notebook 3
      ↓
Data Quality Job
      ↓
Aggregation Job
      ↓
Power BI Refresh
 

Data Engineer have to consider and orchestrate

When does it run?                         什么时候运行？
What are the dependencies?                谁依赖谁？
What is the execution order?              谁先执行？
How are failures handled?                 失败以后怎么办？
How do we handle re-runs / retries?       重新执行怎么办？
How is incremental processing handled?    如何处理增量？
How do we avoid duplicate data?           如何避免重复？
How is data quality enforced/validated?   如何做数据质量？
How is the pipeline monitored?            如何监控？

Declarative Pipeline does this:

Bronze = source data

Silver = cleaned Bronze data

Gold = business aggregation of Silver

------------------------

Silver depends on Bronze

Gold depends on Silver

-----------------------
Bronze
   ↓
Silver
   ↓
Gold

This is called Dependency Graph.

The important architectural shift is: Execution logic becomes separated from data logic. 把 Data Logic 和 Execution Logic 分离。

Your code primarily describes:
(Data Engineers are more focused on: Data Engineer 更关注)

• What data should exist? 数据是什么？
• What transformation defines it? 数据怎么转换？
• What quality rules must it satisfy? 数据质量要求是什么？
• What datasets does it depend on? 数据依赖谁？

The platform can then manage much of (而 Pipeline Framework 更多负责):

• Execution 怎么执行
• Dependency resolution 依赖怎么解析
• Incremental processing 怎么增量更新
• Pipeline updates
• Monitoring 怎么监控
• Data quality metrics
• Recovery 怎么处理失败
• Parallel execution 哪些任务可以并行

                    Declarative Pipeline
                           │
          ┌────────────────┼────────────────┐
          │                │                │
          ↓                ↓                ↓
      Ingestion       Transformation    Data Quality
          │                │                │
     Auto Loader       Spark SQL       Expectations
          │                │                │
          ↓                ↓                ↓
       Bronze  ───────→  Silver  ─────→ Gold
          │
          ↓
      Delta Lake
          │
          ↓
   Unity Catalog

What exactly is the core problem that Declarative Pipelines (DP) aim to solve?

Assuming, a Bank, has those everyday:

Customer
Account
Transaction
Payment
Loan

They come from

Oracle
SQL Server
Kafka
ADLS
REST API

Data Engineer should create:

                Sources
                   │
       ┌───────────┼───────────┐
       ↓           ↓           ↓
    Oracle       Kafka       ADLS
       │           │           │
       └───────────┼───────────┘
                   ↓
                Bronze
                   ↓
                Silver
                   ↓
                 Gold
                   ↓
              Power BI

If we do not have a good pipeline framwork, data engineer may create

ADF Pipeline 1
ADF Pipeline 2
ADF Pipeline 3
Databricks Job 1
Databricks Job 2
Databricks Job 3
Notebook 1
Notebook 2
Notebook 3
DQ Job
CDC Job

maintaining

dependencies
schedules
retries
checkpoints
parameters
error handling
monitoring

One of the primary goals of Declarative Pipelines (DP) is to shift more of this complexity onto the pipeline framework.”

Key foundational concepts of Declarative Pipelines

Declarative Pipelines
│
├── Pipeline
│
├── Dataset
│   ├── Streaming Table
│   ├── Materialized View
│   └── Temporary View
│
├── Flow
│
├── Dependency Graph
│
├── Incremental Processing
│
├── Data Quality
│   └── Expectations
│
├── Change Data Capture
│   └── AUTO CDC
│
├── Batch Processing
│
├── Streaming Processing
│
├── Monitoring
│   └── Event Log
│
├── Unity Catalog
│
└── Deployment / CI-CD

Pipeline / Dataset / Flow

Concept	Professional Meaning	人话
Pipeline	Application / execution boundary	整个工厂
Dataset	Data object produced/maintained	产品
Flow	Data movement / processing path	传送带
Dependency	Relationship between datasets	谁必须等谁
Pipeline Engine	Executes the declared pipeline	工厂管理系统

Declarative Pipeline Python API code implement

# ============================================================
# Import the modern Declarative Pipelines API
# ============================================================

from pyspark import pipelines as dp


# ============================================================
# BRONZE DATASET
# ============================================================
#
# Purpose:
#   Define the Bronze dataset.
#
# Important:
#   We are DECLARING what this dataset should contain.
#   We are not manually starting a streaming query.
# ============================================================

@dp.table
def customer_bronze():

    return (
        spark.readStream

        # ----------------------------------------------------
        # Auto Loader
        #
        # Purpose:
        #   Incrementally discover new files arriving in ADLS.
        # ----------------------------------------------------
        .format("cloudFiles")

        # ----------------------------------------------------
        # Source format
        # ----------------------------------------------------
        .option("cloudFiles.format", "json")

        # ----------------------------------------------------
        # Source location
        # ----------------------------------------------------
        .load("/mnt/raw/customer")
    )


# ============================================================
# SILVER DATASET
# ============================================================
#
# Purpose:
#   Clean and standardize Bronze customer data.
#
# Dependency:
#
#       customer_bronze
#              ↓
#       customer_silver
#
# The dependency is expressed by reading customer_bronze.
# ============================================================

@dp.table
def customer_silver():

    return (
        spark.readStream

        # ----------------------------------------------------
        # Read the Bronze dataset.
        #
        # This establishes the logical dependency:
        #
        # customer_silver
        #        ↓
        # customer_bronze
        # ----------------------------------------------------
        .table("customer_bronze")

        # ----------------------------------------------------
        # Data quality / business rule:
        #
        # A customer record without customer_id is not useful
        # to downstream business processing.
        # ----------------------------------------------------
        .filter("customer_id IS NOT NULL")

        # ----------------------------------------------------
        # Keep only columns required by the Silver layer.
        # ----------------------------------------------------
        .select(
            "customer_id",
            "name",
            "email",
            "updated_at"
        )
    )

Step by Step build a Complete Lakeflow Declarative Pipeline

Customer Data Lakehouse Pipeline — Requirements

1. Project Goal / 项目目标

Build an end-to-end Lakeflow Declarative Pipeline for customer data:

ADLS Landing → Bronze → Data Quality / Routing → Silver → Gold

The pipeline must demonstrate:

• Auto Loader incremental ingestion
• Bronze raw-data preservation
• Data Quality validation
• Invalid-record quarantine
• Severe-invalid record drop with audit logging
• Silver trusted customer data
• Gold business/analytics output
• Declarative Pipeline dependency management
• Incremental processing
• Auditability and traceability

中文

我们建立一个完整的 Lakeflow Declarative Pipeline：

ADLS Landing → Bronze → 数据质量检查/路由 → Silver → Gold

这个实验不是简单地：

读数据 → filter → 写 Silver

而是模拟一个比较接近企业实际的 Data Engineering Pipeline。

重点学习：

• Auto Loader 增量 ingestion
• Bronze 保存原始数据
• Data Quality validation
• Bad data Quarantine
• 严重坏数据 Drop + Audit Log
• Silver Trusted Data
• Gold Analytics Data
• Declarative Pipeline 自动 dependency management
• Incremental processing
• 数据可追溯性和审计

2. Source Data / 数据源

Raw Data:
customer_001.json
customer_002.json
customer_003.json

ADLS Location: abfss://landing@.dfs.core.windows.net/customer/

ADLS
└── landing
    └── customer
        ├── customers_001.json
        ├── customers_002.json
        └── customers_003.json

Our pipeline will implement a simple Medallion Architecture using Lakeflow Declarative Pipelines.

                Azure ADLS
                    │
                    │ JSON files
                    ▼
              Auto Loader
                    │
                    ▼
          ┌───────────────────┐
          │ Bronze             │
          │ customer_bronze    │
          │ Raw / append       │
          └─────────┬─────────┘
                    │
                    │ transformation
                    ▼
          ┌───────────────────┐
          │ Silver            │
          │ customer_silver   │
          │ Cleaned data      │
          └─────────┬─────────┘
                    │
                    │ aggregation
                    ▼
          ┌───────────────────┐
          │ Gold              │
          │ customer_summary  │
          │ Business data     │
          └───────────────────┘

Customer Business Columns

Column	Meaning
customer_id	Customer identifier
name	Customer name
email	Customer email
status	ACTIVE / INACTIVE
country	Customer country
updated_at	Source/business update timestamp

3. Bronze Requirement / Bronze 要求

Bronze 是整个 pipeline 的原始数据入口层。Ingestion 必须使用：Auto Loader. 要能够：support scalable file ingestion
incremental discover new files
process new files
avoid reprocessing already processed files

4. Bronze Must Preserve Raw Data

这是非常重要的一条 requirement。Bronze 不能只保存清洗后的 business columns。Bronze 必须尽可能保留：

Business Data

customer_id
name
email
status
country
updated_at

Raw / Audit Metadata

例如：

_raw_record
_ingest_ts
_source_file
_source_file_name
_source_file_modification_time

如果适用，也可以考虑：

_source_system
_batch_id
_record_hash

5. Important Timestamp Definition

这里要特别区分两个时间, updated_at and _ingest_ts

updated_at: 这是 Source Business Timestamp, 表示源系统认为这条 customer record 什么时候更新。例如： updated_at = 2026-09-10 10:05:00

_ingest_ts: 这是 Data Platform Ingestion Timestamp, 表示这条数据什么时候进入我们的 Data Platform / Pipeline processing。例如：_ingest_ts = 2026-09-10 10:06:13

两者不能混为一谈。

6. Bronze Raw Record Requirement

我们希望 Bronze 能够保存原始 record 的 JSON representation，例如：_raw_record

这样以后：如果一条 record 被 quarantine 或 drop，我们可以看到它进入 pipeline 时的原始状态。这是企业 Data Platform 中非常重要的：Traceability / 可追溯性

7. Data Quality Requirements

这是整个实验最重要的部分。我们定义三个核心 Data Quality rules：

Rule 1 — Customer ID

customer_id must not be NULL

Rule 2 — Email

Email 必须：

NOT NULL

并且应该满足基本 email format。例如：alice@example.com 有效。而：alice@ 无效。

Rule 3 — Status

Status 必须是：ACTIVE 或者：INACTIVE . 其他值都是 invalid。 例如：

ACTIVE       → valid
INACTIVE     → valid
PENDING      → invalid
UNKNOWN      → invalid
NULL         → invalid

8. Data Quality Routing Requirement

我们不是简单地 DROP 所有坏数据。这是这个项目的核心设计。我们把 record 分成三个类别：

                    Bronze
                       │
                       ▼
                Data Quality
                   Validation
                       │
          ┌────────────┼────────────┐
          ▼            ▼            ▼
        VALID      RECOVERABLE    SEVERE
          │            │            │
          ▼            ▼            ▼
       Silver      Quarantine    Drop Log

9. Routing Decision Matrix

这是目前我们确认的正式 requirement baseline。

Customer_ID	Email	Status	Action	Destination
Valid	Valid	ACTIVE	ACCEPT	Silver
Valid	Valid	INACTIVE	ACCEPT	Silver
NULL	Valid	Valid	QUARANTINE	Quarantine
Valid	NULL	Valid	QUARANTINE	Quarantine
Valid	Valid	Invalid	QUARANTINE	Quarantine
NULL	NULL	Valid	QUARANTINE	Quarantine
NULL	Valid	Invalid	QUARANTINE	Quarantine
Valid	NULL	Invalid	QUARANTINE	Quarantine
NULL	NULL	Invalid	DROP	Drop Log

10. Quarantine Requirement

如果 record 不符合正常 Silver 数据质量要求，但是没有达到 Severe Drop 条件：Quarantine

例如：

customer_id = NULL
email = bob@example.com
status = ACTIVE

不能进入 Silver。但是：不能直接删除。应该进入：

customer_quarantine

11. Quarantine Must Preserve Original Record

Quarantine 至少应该包含：Original Business Data

customer_id
name
email
status
country
updated_at

Original Record

_raw_record

Source Information

_source_file
_source_file_name

Ingestion Information

_ingest_ts

Data Quality Information

_dq_customer_id_valid
_dq_email_valid
_dq_status_valid

以及：

_dq_reason

例如：

CUSTOMER_ID_NULL

或者：

EMAIL_NULL

或者：

INVALID_STATUS

多个问题：

CUSTOMER_ID_NULL;EMAIL_NULL

12. Quarantine Timestamp

需要记录：_quarantine_ts . 表示：这条 record 被判定并进入 quarantine 的时间。

13. Drop Requirement

最严重的情况：

customer_id = NULL
email       = NULL
status      = INVALID

这条 record：不进入 Silver, 不作为正常 Quarantine Record. 而是：DROP

但是：Drop ≠ 数据完全消失,

这是非常重要的概念。真正的设计是：

Bronze
   │
   └── severe invalid
             │
             ▼
        Drop Audit Log

所以我们实际上是：从 business processing path 中 drop，但从 audit trail 中保留。

14. Drop Log Requirement

建立：customer_drop_log

至少保存：

Original Data

customer_id
name
email
status
country
updated_at

Raw Record

_raw_record

Source Metadata

_source_file
_source_file_name
_source_file_modification_time

Ingestion Information

_ingest_ts

Drop Information

_drop_ts
_drop_reason

例如：

_drop_reason =
CUSTOMER_ID_NULL;EMAIL_NULL;INVALID_STATUS

这样以后可以回答：Why was this customer record dropped?

以及：What exactly was the original record?

15. Silver Requirement

Silver 是：Trusted / Validated Customer Data

只有：

customer_id valid
AND
email valid
AND
status ∈ ('ACTIVE', 'INACTIVE')

的 record 才能进入 Silver。因此：

Bronze
   │
   ├── valid ─────────────→ Silver
   │
   ├── invalid/recoverable → Quarantine
   │
   └── severe invalid ────→ Drop Log

Silver 不应该包含：

• NULL customer_id
• NULL email
• invalid email
• invalid status

16. Silver Transformation Requirement

Silver 除了 DQ filtering，还可以做一些正常的 standardization，例如：

trim(customer_id)
trim(name)
lower(email)
upper(status)

例如：

" alice@example.com "

standardize 成：

"alice@example.com"

但是：Bronze 不修改原始数据。这是：

Bronze = Preserve
Silver = Clean / Standardize / Trust

17. Gold Requirement

Gold 是面向：Business / Analytics Consumption. 我们需要一个简单但有意义的 Gold dataset。

例如：

customer_summary

按：

status

统计 customer 数量：

status      customer_count
---------------------------
ACTIVE      5
INACTIVE    1

也可以进一步按：

country
status

进行 aggregation。

例如：

country    status       customer_count
---------------------------------------
Canada     ACTIVE       4
Canada     INACTIVE     0
USA        INACTIVE     1
UK         ACTIVE       1

Gold 使用：Materialized View 来体现：

Silver
   ↓
Gold Materialized View

18. Final Pipeline Architecture

                         ADLS
                          │
                          │
                    Auto Loader
                          │
                          ▼
                    ┌───────────┐
                    │  Bronze   │
                    │ Raw Data  │
                    │ + Metadata│
                    └─────┬─────┘
                          │
                          ▼
                  Data Quality Rules
                          │
             ┌────────────┼────────────┐
             │            │            │
             ▼            ▼            ▼
          VALID      RECOVERABLE     SEVERE
             │            │            │
             ▼            ▼            ▼
          Silver      Quarantine    Drop Log
             │
             │
             ▼
       Gold Materialized
             View

19. Final Dataset List

整个 Pipeline 最终应该至少有 4 个主要数据集：

1. customer_bronze
2. customer_silver
3. customer_quarantine
4. customer_drop_log
5. customer_summary        ← Gold

也就是说：

Bronze
   │
   ├──────────────→ Silver
   │                   │
   │                   ▼
   │                  Gold
   │
   ├──────────────→ Quarantine
   │
   └──────────────→ Drop Log

20. Declarative Pipeline Requirement

我们不希望自己手工 orchestration, 而是 Declare the datasets and their relationships.

Silver reads Bronze
Gold reads Silver
Quarantine reads Bronze
Drop Log reads Bronze

Pipeline framework 根据这些 dependencies 自动建立：

Dependency Graph

然后负责：

• execution
• dependency resolution
• orchestration
• incremental processing
• checkpointing
• pipeline updates
• table maintenance

我们的代码主要负责：定义“数据应该是什么样子，以及它从哪里来”。

22. Incremental Processing Requirement

实验必须体现：First Run 例如：customers_001.json ; customers_002.json, 进入 pipeline。

Later 再增加：customers_003.json

Pipeline update 后：只处理新的 input data，而不是每次重新从头读取所有文件。

最终我们观察：

Bronze
Silver
Quarantine
Drop Log
Gold

23. Data Lineage / Traceability Requirement

我们最终应该能够回答：Where did this Silver record come from?

Silver
 ↓
Bronze
 ↓
ADLS file

Why was this record quarantined?

Quarantine
 ↓
DQ reason
 ↓
Original record
 ↓
Source file

Why was this record dropped?

Drop Log
 ↓
Drop reason
 ↓
Original record
 ↓
Source file
 ↓
Ingestion timestamp

这就是：

Data Lineage + Auditability + Traceability

25. Coding Order / 我们之后的 Coding 顺序

Step 1: Create Bronze

ADLS
 ↓
Auto Loader
 ↓
customer_bronze

Step 2: Validate Bronze Data

建立：

customer_id validation
email validation
status validation

Step 3: Classify / Route Records

VALID
QUARANTINE
DROP

Step 4: Create Quarantine

customer_quarantine

Step 5: Create Drop Audit Log

customer_drop_log

Step 6: Create Silver

customer_silver

Step 7: Create Gold

customer_summary

Step 8: Test Incremental Processing

增加新的 JSON：

customers_003.json

Step 9: Review Pipeline Graph / Data Quality / Audit

Test Data

we want to demonstrate incremental ingestion.

# File 1 — customers_001.json

[
  {
    "customer_id": "C001",
    "name": "Alice",
    "email": "alice@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:00:00"
  },
  {
    "customer_id": "C002",
    "name": "Bob",
    "email": "bob@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:05:00"
  },
  {
    "customer_id": "C003",
    "name": "Charlie",
    "email": "charlie@example.com",
    "status": "INACTIVE",
    "country": "USA",
    "updated_at": "2026-09-10T09:10:00"
  }
]





# File 2 — customers_002.json
[
  {
    "customer_id": "C004",
    "name": "David",
    "email": "david@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T10:00:00"
  },
  {
    "customer_id": "C005",
    "name": "Emma",
    "email": "emma@example.com",
    "status": "ACTIVE",
    "country": "UK",
    "updated_at": "2026-09-10T10:05:00"
  }
]

URL: abfss://landing@storageaccount.dfs.core.windows.net/customer/

Pipeline source code will like:

customer_pipeline/
│
├── customer_pipeline.py
│
└── configuration

1: Create Bronze (Simple, no for product)

aim: Incrementally load JSON files from ADLS into a Bronze Streaming Table.

# 简单实现， 其实在实际项目不会这么简单
from pyspark import pipelines as dp


@dp.table 
def customer_bronze():        # create a table called "customer_bronze" 
   # The decorator declares that the function defines a pipeline-managed dataset.
   # In this context, it creates a streaming table.

    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load("abfss://landing@myaccount.dfs.core.windows.net/customer/")
    )

we do not do ” create table ……” then “writeStream.start()” ,

Declarative Pipeline WILL TAKE OVER this job

The pipeline framework manages:

• execution
• checkpointing
• dependency resolution
• incremental processing
• scheduling
• table maintenance
• orchestration

The developer primarily declares: what dataset should exist and how it is derived.

2: Create Silver (Simple, not for product)

@dp.table
@dp.expect(                          # Data Quality Expectation
    "valid_customer_id",
    "customer_id IS NOT NULL"        # customer_id cannot null
)

@dp.expect(
    "valid_email",
    "email IS NOT NULL AND email RLIKE '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'"
                                           # email cannot be null and it should be in good form
)

@dp.expect(
    "valid_status",
    "status IN ('ACTIVE', 'INACTIVE')"    # status only allow ACTIVE or INACTIVE
)

def customer_silver():

    return (
        spark.readStream
            .table("customer_bronze")
            .filter("customer_id IS NOT NULL")
            .select(
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at"
            )
    )

Product grade, end- to-end code implement

test data

abfss://landing@<storage-account>.dfs.core.windows.net/customer/

customers_001.json
customers_002.json

[
  {
    "customer_id": "C001",
    "name": "Alice",
    "email": "alice@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:00:00"
  },
  {
    "customer_id": "C002",
    "name": "Bob",
    "email": "bob@example.com",
    "status": "ACTIVE",
    "country": "Canada",
    "updated_at": "2026-09-10T09:05:00"
  },
  {
    "customer_id": "C003",
    "name": "Charlie",
    "email": "charlie@example.com",
    "status": "INACTIVE",
    "country": "USA",
    "updated_at": "2026-09-10T09:10:00"
  }
]

Step 1 — Create Bronze

Our first pipeline layer is:

ADLS Landing
      │
      │ Auto Loader
      ▼
customer_bronze

Bronze’s job is not to clean the data. Its job is: Ingest and preserve the source data faithfully, together with ingestion/source metadata.

First Bronze implementation

from pyspark import pipelines as dp


@dp.table
def customer_bronze():  # create a table called "customer_bronze" 

    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load(
                "abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
            )
    )

What does this code actually produce? it creates “customer_bronze”

Table customer_bronze:

customer_id
name
email
status
country
updated_at

# we're missing
├── _raw_record
├── _ingest_ts
├── _source_file
├── _source_file_name
└── _source_file_modification_time
这些是我们之前 Requirement 中要求的 audit information。只有改成下面的

现在我们把代码升级， 满足requirement 的要求。

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    current_timestamp,
    col,
    struct,
    to_json
)


@dp.table(
    comment="Bronze layer containing raw customer data ingested incrementally from ADLS using Auto Loader."
)
def customer_bronze():

    return (
        spark.readStream

            # ---------------------------------------------------------
            # Use Databricks Auto Loader as the streaming file source.
            # Auto Loader incrementally discovers new files arriving
            # in the ADLS landing folder.
            # ---------------------------------------------------------
            .format("cloudFiles")

            # ---------------------------------------------------------
            # The source files in our experiment are JSON files.
            # ---------------------------------------------------------
            .option("cloudFiles.format", "json")

            # ---------------------------------------------------------
            # Load files from the ADLS landing zone.
            #
            # Replace <storage-account> with the real storage account.
            # ---------------------------------------------------------
            .load(
                "abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
            )

            # ---------------------------------------------------------
            # Add platform ingestion timestamp.
            #
            # This is NOT the source business timestamp.
            # updated_at = source/business time
            # _ingest_ts = data-platform ingestion/processing time
            # ---------------------------------------------------------
            .withColumn(
                "_ingest_ts",
                current_timestamp()
            )

            # ---------------------------------------------------------
            # Capture source file metadata.
            #
            # _metadata is provided by Auto Loader.
            # We preserve the file path and modification time so that
            # downstream DQ issues can be traced back to the source file.
            # ---------------------------------------------------------
            .withColumn(
                "_source_file",
                col("_metadata.file_path")
            )
            .withColumn(
                "_source_file_modification_time",
                col("_metadata.file_modification_time")
            )

            # ---------------------------------------------------------
            # Keep the raw source record as JSON.
            #
            # This gives us a representation of the record as it exists
            # at the Bronze stage before Silver transformations.
            #
            # This will later be extremely useful for:
            #   - Quarantine
            #   - Drop Audit Log
            #   - Troubleshooting
            #   - Data lineage / traceability
            # ---------------------------------------------------------
            .withColumn(
                "_raw_record",
                to_json(
                    struct(
                        "customer_id",
                        "name",
                        "email",
                        "status",
                        "country",
                        "updated_at"
                    )
                )
            )

            # ---------------------------------------------------------
            # Select the columns that belong to the Bronze contract.
            #
            # Bronze preserves business data plus source/audit metadata.
            # ---------------------------------------------------------
            .select(
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",

                "_raw_record",
                "_ingest_ts",

                "_source_file",
                "_source_file_modification_time"
            )
    )

三个时间的区别 (以后你面试很可能会被问到)

Field	Meaning
updated_at	Source business timestamp
_source_file_modification_time	Source file timestamp
_ingest_ts	Platform ingestion timestamp

为什么需要 _raw_record？ 这是我们后面设计：

Quarantine
Drop Log

Current Bronze Architecture

现在我们已经从：

ADLS
 ↓
Auto Loader
 ↓
Bronze

升级成：

                         ADLS
                          │
                          ▼
                    Auto Loader
                          │
                          ▼
                ┌──────────────────┐
                │ customer_bronze  │
                ├──────────────────┤
                │ customer_id      │
                │ name             │
                │ email            │
                │ status           │
                │ country          │
                │ updated_at       │
                │                  │
                │ _raw_record      │
                │ _ingest_ts       │
                │ _source_file     │
                │ _source_file_... │
                └──────────────────┘

What should we see after the first run? 如果 source 中：

C001
C002
C003
C004
C005

那么 Bronze 应该有：5 records. 例如：

customer_id	email	status	_source_file	_ingest_ts
C001	alice@example.com	ACTIVE	customers_001.json	…
C002	bob@example.com	ACTIVE	customers_001.json	…
C003	charlie@example.com	INACTIVE	customers_001.json	…
C004	david@example.com	ACTIVE	customers_002.json	…
C005	emma@example.com	ACTIVE	customers_002.json	…

注意：我们还没有进行任何 Business Data Quality filtering。

所以把代码补完整：

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    current_timestamp,
    col,
    struct,
    to_json
)

@dp.table(
    comment="Bronze layer containing raw customer data ingested incrementally from ADLS using Auto Loader."
)
def customer_bronze():

    return (
        spark.readStream
            .format("cloudFiles")
            .option("cloudFiles.format", "json")
            .load(
                "abfss://landing@<storage-account>.dfs.core.windows.net/customer/"
            )

            # Platform ingestion timestamp
            .withColumn(
                "_ingest_ts",
                current_timestamp()
            )

            # Source file metadata
            .withColumn(
                "_source_file",
                col("_metadata.file_path")
            )
            .withColumn(
                "_source_file_name",
                col("_metadata.file_name")
            )
            .withColumn(
                "_source_file_modification_time",
                col("_metadata.file_modification_time")
            )

            # Serialized representation of the parsed business record
            .withColumn(
                "_raw_record",
                to_json(
                    struct(
                        "customer_id",
                        "name",
                        "email",
                        "status",
                        "country",
                        "updated_at"
                    )
                )
            )

            .select(
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",
                "_raw_record",
                "_ingest_ts",
                "_source_file",
                "_source_file_name",
                "_source_file_modification_time"
            )
    )

Step 2 — Data Quality Validation

数据质量验证

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    col,
    when,
    concat_ws,
    array,
    lit
)

@dp.temporary_view
def customer_validation():

    return (
        spark.readStream
            .table("customer_bronze")

            # ==============================
            # DQ Rule 1: Customer ID
            # ==============================
            .withColumn(
                "_dq_customer_id_valid",
                col("customer_id").isNotNull()
            )

            # ==============================
            # DQ Rule 2: Email
            # ==============================
            .withColumn(
                "_dq_email_valid",
                (
                    col("email").isNotNull()
                    &
                    col("email").rlike(
                        r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                    )
                )
            )

            # ==============================
            # DQ Rule 3: Status
            # ==============================
            .withColumn(
                "_dq_status_valid",
                col("status").isin("ACTIVE", "INACTIVE")
            )

            # ==============================
            # Overall validation
            # ==============================
            .withColumn(
                "_dq_all_valid",
                (
                    col("_dq_customer_id_valid")
                    &
                    col("_dq_email_valid")
                    &
                    col("_dq_status_valid")
                )
            )

            # ==============================
            # DQ reason
            # ==============================
            .withColumn(
                "_dq_reason",
                concat_ws(
                    ";",
                    array(
                        when(
                            col("customer_id").isNull(),
                            lit("CUSTOMER_ID_NULL")
                        ),

                        when(
                            col("email").isNull(),
                            lit("EMAIL_NULL")
                        ),

                        when(
                            col("email").isNotNull()
                            &
                            ~col("email").rlike(
                                r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                            ),
                            lit("INVALID_EMAIL_FORMAT")
                        ),

                        when(
                            ~col("status").isin("ACTIVE", "INACTIVE"),
                            lit("INVALID_STATUS")
                        )
                    )
                )
            )

            # ==============================
            # Routing decision
            # ==============================
            .withColumn(
                "_dq_action",
                when(
                    col("_dq_all_valid"),
                    lit("ACCEPT")
                )
                .when(
                    (~col("_dq_customer_id_valid"))
                    &
                    (~col("_dq_email_valid"))
                    &
                    (~col("_dq_status_valid")),
                    lit("DROP")
                )
                .otherwise(
                    lit("QUARANTINE")
                )
            )
    )

Step 3 — Routing into Silver / Quarantine / Drop Log

Silver 的原则：Only trusted, validated customer records enter Silver.

from pyspark import pipelines as dp
from pyspark.sql.functions import (
    col,
    when,
    concat_ws,
    array,
    lit,
    trim,
    lower,
    upper,
    current_timestamp
)


# ============================================================
# 1. Validation and Routing
# ============================================================

@dp.temporary_view
def customer_validation():

    return (
        spark.readStream
            .table("customer_bronze")

            # ------------------------------------------------
            # DQ Rule 1: customer_id must not be NULL
            # ------------------------------------------------
            .withColumn(
                "_dq_customer_id_valid",
                col("customer_id").isNotNull()
            )

            # ------------------------------------------------
            # DQ Rule 2:
            # email must not be NULL and must match
            # the basic email format
            # ------------------------------------------------
            .withColumn(
                "_dq_email_valid",
                (
                    col("email").isNotNull()
                    &
                    col("email").rlike(
                        r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                    )
                )
            )

            # ------------------------------------------------
            # DQ Rule 3:
            # status must be ACTIVE or INACTIVE
            # ------------------------------------------------
            .withColumn(
                "_dq_status_valid",
                (
                    col("status").isNotNull()
                    &
                    col("status").isin(
                        "ACTIVE",
                        "INACTIVE"
                    )
                )
            )

            # ------------------------------------------------
            # Overall DQ result
            # ------------------------------------------------
            .withColumn(
                "_dq_all_valid",
                (
                    col("_dq_customer_id_valid")
                    &
                    col("_dq_email_valid")
                    &
                    col("_dq_status_valid")
                )
            )

            # ------------------------------------------------
            # DQ Reason
            #
            # Multiple reasons are concatenated:
            #
            # CUSTOMER_ID_NULL;EMAIL_NULL
            # ------------------------------------------------
            .withColumn(
                "_dq_reason",
                concat_ws(
                    ";",
                    array(

                        when(
                            col("customer_id").isNull(),
                            lit("CUSTOMER_ID_NULL")
                        ),

                        when(
                            col("email").isNull(),
                            lit("EMAIL_NULL")
                        ),

                        when(
                            col("email").isNotNull()
                            &
                            ~col("email").rlike(
                                r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$"
                            ),
                            lit("INVALID_EMAIL_FORMAT")
                        ),

                        when(
                            col("status").isNull()
                            |
                            ~col("status").isin(
                                "ACTIVE",
                                "INACTIVE"
                            ),
                            lit("INVALID_STATUS")
                        )
                    )
                )
            )

            # ------------------------------------------------
            # Routing Decision
            #
            # ACCEPT:
            #   All three rules pass
            #
            # DROP:
            #   All three rules fail
            #
            # QUARANTINE:
            #   Everything else
            # ------------------------------------------------
            .withColumn(
                "_dq_action",
                when(
                    col("_dq_all_valid"),
                    lit("ACCEPT")
                )
                .when(
                    (~col("_dq_customer_id_valid"))
                    &
                    (~col("_dq_email_valid"))
                    &
                    (~col("_dq_status_valid")),
                    lit("DROP")
                )
                .otherwise(
                    lit("QUARANTINE")
                )
            )
    )


# ============================================================
# 2. Silver
# ============================================================

@dp.table(
    comment="Trusted customer records that passed all core data quality rules."
)
def customer_silver():

    return (
        spark.readStream
            .table("customer_validation")

            # Only fully valid records
            .filter(
                col("_dq_action") == "ACCEPT"
            )

            # Standardize trusted business data
            .select(
                trim(
                    col("customer_id")
                ).alias("customer_id"),

                trim(
                    col("name")
                ).alias("name"),

                lower(
                    trim(
                        col("email")
                    )
                ).alias("email"),

                upper(
                    trim(
                        col("status")
                    )
                ).alias("status"),

                trim(
                    col("country")
                ).alias("country"),

                col("updated_at"),

                # Audit / lineage metadata
                col("_raw_record"),
                col("_ingest_ts"),
                col("_source_file"),
                col("_source_file_name"),
                col("_source_file_modification_time")
            )
    )


# ============================================================
# 3. Quarantine
# ============================================================

@dp.table(
    comment="Customer records that failed one or more recoverable data quality rules."
)
def customer_quarantine():

    return (
        spark.readStream
            .table("customer_validation")

            # Recoverable DQ failures
            .filter(
                col("_dq_action") == "QUARANTINE"
            )

            .select(
                # Original business data
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",

                # Raw / lineage information
                "_raw_record",
                "_ingest_ts",
                "_source_file",
                "_source_file_name",
                "_source_file_modification_time",

                # DQ results
                "_dq_customer_id_valid",
                "_dq_email_valid",
                "_dq_status_valid",
                "_dq_reason",

                # Quarantine audit timestamp
                current_timestamp().alias(
                    "_quarantine_ts"
                )
            )
    )


# ============================================================
# 4. Drop Log
# ============================================================

@dp.table(
    comment="Audit log for severely invalid customer records excluded from normal processing."
)
def customer_drop_log():

    return (
        spark.readStream
            .table("customer_validation")

            # Severe DQ failures
            .filter(
                col("_dq_action") == "DROP"
            )

            .select(
                # Original business data
                "customer_id",
                "name",
                "email",
                "status",
                "country",
                "updated_at",

                # Raw / lineage information
                "_raw_record",
                "_ingest_ts",
                "_source_file",
                "_source_file_name",
                "_source_file_modification_time",

                # DQ results
                "_dq_customer_id_valid",
                "_dq_email_valid",
                "_dq_status_valid",

                # Audit reason
                col("_dq_reason").alias(
                    "_drop_reason"
                ),

                # Drop audit timestamp
                current_timestamp().alias(
                    "_drop_ts"
                )
            )
    )

这四个 Dataset 的关系

                    customer_bronze
                           │
                           │
                           ▼
                ┌────────────────────┐
                │ customer_validation│
                │                    │
                │ DQ Rules           │
                │ + Reason           │
                │ + Routing          │
                └─────────┬──────────┘
                          │
             ┌────────────┼────────────┐
             │            │            │
          ACCEPT      QUARANTINE      DROP
             │            │            │
             ▼            ▼            ▼
     customer_silver   customer_     customer_
                      quarantine    drop_log

最终 Routing Logic

customer_id	email	status	_dq_action	去哪里
valid	valid	ACTIVE	ACCEPT	Silver
valid	valid	INACTIVE	ACCEPT	Silver
NULL	valid	valid	QUARANTINE	Quarantine
valid	NULL	valid	QUARANTINE	Quarantine
valid	invalid format	valid	QUARANTINE	Quarantine
valid	valid	invalid	QUARANTINE	Quarantine
NULL	NULL	valid	QUARANTINE	Quarantine
NULL	valid	invalid	QUARANTINE	Quarantine
valid	NULL	invalid	QUARANTINE	Quarantine
NULL	NULL	invalid	DROP	Drop Log

Step 4 — Gold: customer_summary

from pyspark import pipelines as dp


@dp.materialized_view(
    comment="Gold customer summary by customer status."
)
def customer_summary():

    return (
        spark.read
            .table("customer_silver")
            .groupBy("status")
            .count()
            .withColumnRenamed(
                "count",
                "customer_count"
            )
    )

完整架构：

                       ADLS Landing
                            │
                            ▼
                     ┌─────────────┐
                     │ Auto Loader │
                     └──────┬──────┘
                            │
                            ▼
                    ┌───────────────┐
                    │    Bronze     │
                    │ Raw + Metadata│
                    └───────┬───────┘
                            │
                            ▼
                 ┌─────────────────────┐
                 │ Customer Validation │
                 │                     │
                 │ DQ Rules            │
                 │ DQ Reason           │
                 │ Routing             │
                 └─────────┬───────────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
           ACCEPT      QUARANTINE      DROP
              │            │            │
              ▼            ▼            ▼
           Silver      Quarantine    Drop Log
              │
              ▼
        Materialized View
              │
              ▼
             Gold
              │
              ▼
     customer_summary

Lakeflow Declarative Pipeline

我们现在写的是：

@dp.table
def customer_bronze():
    ...

然后：

@dp.temporary_view
def customer_validation():
    return spark.readStream.table("customer_bronze")

然后：

@dp.table
def customer_silver():
    return spark.readStream.table("customer_validation")

最后：

@dp.materialized_view
def customer_summary():
    return spark.read.table("customer_silver")

Streaming Table vs Materialized View

	Streaming Table	Materialized View
典型用途	incremental row processing	derived / aggregated result
数据特点	持续到来的数据	查询计算结果
常见操作	filter / transform / enrich	aggregation / business metrics
API	@dp.table	@dp.materialized_view
我们的例子	customer_silver	customer_summary

Step 6 — Put the Code into a Lakeflow Pipeline

现在进入真正的 hands-on 部分：

我们已经写完 Dataset definitions。下一步就是把这些 definitions 放进一个 Lakeflow Pipeline，让 Databricks 真正执行它。

这里先只做 Pipeline structure + code organization，暂时不深入 Deployment / Schedule / Production settings。

先把代码组织起来

最简单的方式是一个 Python pipeline source file，例如：

customer_pipeline/
│
└── customer_pipeline.py

里面包含我们目前的全部 Dataset definitions：

customer_pipeline.py
│
├── customer_bronze
│
├── customer_validation
│
├── customer_silver
│
├── customer_quarantine
│
├── customer_drop_log
│
└── customer_summary