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
↓
GoldTraditional 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
↓
GoldThis 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 CatalogWhat exactly is the core problem that Declarative Pipelines (DP) aim to solve?
Assuming, a Bank, has those everyday:
Customer
Account
Transaction
Payment
LoanThey come from
Oracle
SQL Server
Kafka
ADLS
REST APIData Engineer should create:
Sources
│
┌───────────┼───────────┐
↓ ↓ ↓
Oracle Kafka ADLS
│ │ │
└───────────┼───────────┘
↓
Bronze
↓
Silver
↓
Gold
↓
Power BIIf 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 Jobmaintaining
dependencies
schedules
retries
checkpoints
parameters
error handling
monitoringOne 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-CDPipeline / 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.jsonOur 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 |
| 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_atRaw / Audit Metadata
例如：
_raw_record
_ingest_ts
_source_file
_source_file_name
_source_file_modification_time如果适用，也可以考虑：
_source_system
_batch_id
_record_hash5. 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 NULLRule 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 → invalid8. Data Quality Routing Requirement
我们不是简单地 DROP 所有坏数据。这是这个项目的核心设计。我们把 record 分成三个类别：
Bronze
│
▼
Data Quality
Validation
│
┌────────────┼────────────┐
▼ ▼ ▼
VALID RECOVERABLE SEVERE
│ │ │
▼ ▼ ▼
Silver Quarantine Drop Log9. Routing Decision Matrix
这是目前我们确认的正式 requirement baseline。
| Customer_ID | 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_quarantine11. Quarantine Must Preserve Original Record
Quarantine 至少应该包含：Original Business Data
customer_id
name
email
status
country
updated_atOriginal Record
_raw_recordSource Information
_source_file
_source_file_nameIngestion Information
_ingest_tsData 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_NULL12. 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_atRaw Record
_raw_recordSource Metadata
_source_file
_source_file_name
_source_file_modification_timeIngestion Information
_ingest_tsDrop 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 LogSilver 不应该包含：
- 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 / Trust17. 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 1Gold 使用：Materialized View 来体现：
Silver
↓
Gold Materialized View18. Final Pipeline Architecture
ADLS
│
│
Auto Loader
│
▼
┌───────────┐
│ Bronze │
│ Raw Data │
│ + Metadata│
└─────┬─────┘
│
▼
Data Quality Rules
│
┌────────────┼────────────┐
│ │ │
▼ ▼ ▼
VALID RECOVERABLE SEVERE
│ │ │
▼ ▼ ▼
Silver Quarantine Drop Log
│
│
▼
Gold Materialized
View19. 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 Log20. Declarative Pipeline Requirement
我们不希望自己手工 orchestration, 而是 Declare the datasets and their relationships.
Silver reads Bronze
Gold reads Silver
Quarantine reads Bronze
Drop Log reads BronzePipeline 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
Gold23. Data Lineage / Traceability Requirement
我们最终应该能够回答：Where did this Silver record come from?
Silver
↓
Bronze
↓
ADLS fileWhy was this record quarantined?
Quarantine
↓
DQ reason
↓
Original record
↓
Source fileWhy 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_bronzeStep 2: Validate Bronze Data
建立：
customer_id validation
email validation
status validationStep 3: Classify / Route Records
VALID
QUARANTINE
DROPStep 4: Create Quarantine
customer_quarantineStep 5: Create Drop Audit Log
customer_drop_logStep 6: Create Silver
customer_silverStep 7: Create Gold
customer_summaryStep 8: Test Incremental Processing
增加新的 JSON：
customers_003.jsonStep 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_bronzeBronze’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 LogCurrent 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 | 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 | 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_summaryLakeflow 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