databricks-pipelines

Lakeflow Spark 声明式管道开发

首先：使用父级 databricks-core 技能了解 CLI 基础知识、身份验证、配置文件选择和数据发现命令。

决策树

使用此树确定要使用的数据集类型和功能。同一数据集可以应用多种功能——例如，流式表可以使用 Auto Loader 进行摄取，使用 Append Flows 扇入，并使用 Expectations 保证数据质量。先选择数据集类型，再叠加适用的功能。

User request → What kind of output?
├── Intermediate/reusable logic (not persisted) → Temporary View
│   ├── Preprocessing/filtering before Auto CDC → Temporary View feeding CDC flow
│   ├── Shared intermediate streaming logic reused by multiple downstream tables
│   ├── Pipeline-private helper logic (not published to catalog)
│   └── Published to UC for external queries → Persistent View (SQL only)
├── Persisted dataset
│   ├── Source is a periodic complete snapshot that must be diffed by key → Streaming Table + Auto CDC FROM SNAPSHOT (SQL requires DBR 18.3+)
│   ├── Source is streaming/incremental/continuously growing → Streaming Table
│   │   ├── File ingestion (cloud storage, Volumes) → Auto Loader
│   │   ├── Message bus (Kafka, Kinesis, Pub/Sub, Pulsar, Event Hubs) → streaming source read
│   │   ├── Existing streaming/Delta table → streaming read from table
│   │   ├── CDC / upserts / track changes / keep latest per key / SCD Type 1 or 2 → Auto CDC
│   │   ├── Multiple sources into one table → Append Flows (NOT union)
│   │   ├── Historical backfill + live stream → one-time Append Flow + regular flow
│   │   └── Windowed aggregation with watermark → stateful streaming
│   └── Source is batch/historical/full scan → Materialized View
│       ├── Aggregation/join across full dataset (GROUP BY, SUM, COUNT, etc.)
│       ├── Gold layer aggregation from streaming table → MV with batch read (spark.read / no STREAM)
│       ├── JDBC/Federation/external batch sources
│       └── Small static file load (reference data, no streaming read)
├── Output to external system (Python only) → Sink
│   ├── Existing external table not managed by this pipeline → Sink with format="delta"
│   │   (prefer fully-qualified dataset names if the pipeline should own the table — see Publishing Modes)
│   ├── Kafka / Event Hubs → Sink with format="kafka" + @dp.append_flow(target="sink_name")
│   ├── Custom destination not natively supported → Sink with custom format
│   ├── Custom merge/upsert logic per batch → ForEachBatch Sink (Public Preview)
│   └── Multiple destinations per batch → ForEachBatch Sink (Public Preview)
└── Data quality constraints → Expectations (on any dataset type)

常见陷阱

• 名称 → SDP = LDP = Lakeflow Declarative Pipelines =（以前称为）DLT。用户提到这些名称时，它们可以互换使用。
• 未指定类型的“创建表” → 询问数据源是流式还是批处理。流式数据源 → 流式表；批处理数据源 → 物化视图。不匹配的组合会在验证时出错。
• 对流式数据源进行聚合 → 使用通过批处理读取的物化视图（spark.read.table / 不带 STREAM 的 SELECT FROM）。ST 仅支持追加，当源行发生变化时不会重新计算聚合；MV 会重新计算。
• 中间逻辑 → 默认使用临时视图。即使多个下游表会复用共享逻辑，也应如此。仅当物化一次能显著减少重复处理时，才使用私有 MV/ST（private=True / CREATE PRIVATE ...）。在 Auto CDC 前进行预处理时，必须使用临时视图——CDC 流会从 STREAM(view_name)（SQL）或 spark.readStream.table("view_name")（Python）读取。
• 合并流 → 使用多个 Append Flows。跨流式数据源执行 UNION 是反模式。
• 更改数据集类型 → 无法原地将 ST 更改为 MV，或将 MV 更改为 ST。完全刷新也无济于事。手动删除现有表，或重命名新数据集。
• CREATE OR REFRESH 与 CREATE → 两者都能解析 SQL 数据集，但 CREATE OR REFRESH 是惯用写法。对于 PRIVATE 数据集：CREATE OR REFRESH PRIVATE STREAMING TABLE / ... MATERIALIZED VIEW。
• Kafka/Event Hubs 接收器序列化 → 必须包含 value 列；使用 to_json(struct(*)) AS value 序列化行。请参阅 sink-python.md。
• 多列 Auto CDC 排序 → SQL：SEQUENCE BY STRUCT(col1, col2)。Python：sequence_by=struct("col1", "col2")。请参阅 Auto CDC 参考文档。
• Auto CDC TRUNCATE（仅限 SCD Type 1）→ SQL：APPLY AS TRUNCATE WHEN condition。Python：apply_as_truncates=expr("condition")。不要声称不支持 truncate。
• 定期完整快照 → 在 DBR 18.3+ 上使用 SQL AUTO CDC FROM SNAPSHOT，或在较早的运行时上使用 Python dp.create_auto_cdc_from_snapshot_flow。请参阅 Auto CDC 参考文档，了解版本选择和生命周期规则。
• 仅限 Python 的功能 → Sinks、ForEachBatch Sinks 和自定义数据源仅支持 Python。用户使用 SQL 时，请说明这一点，并建议切换到 Python。
• 推荐一种明确方案 → 提供一条推荐路径。不要罗列反模式或较差的替代方案，以免造成困惑。只有在替代方案确实存在不同权衡时，才提及它们。

常见问题

Agent 经常遇到的错误 → 原因/修复方法。有关 DAB 捆绑包与 CLI 迭代部署问题，请参阅特定工作流的参考文件。

错误 / 症状	原因 / 修复方法
拒绝 CREATE OR REPLACE STREAMING TABLE / MATERIALIZED VIEW	CREATE OR REPLACE 是标准 SQL，不是 SDP。请使用 CREATE OR REFRESH STREAMING TABLE / CREATE OR REFRESH MATERIALIZED VIEW。
CLI 对 databricks fs ls /Volumes/... 报错	即使是 UC Volume 路径，也必须带上 dbfs: 前缀：databricks fs ls dbfs:/Volumes/<catalog>/<schema>/<volume>/<path>。
首次写入时出现 DELTA_CLUSTERING_COLUMNS_DATATYPE_NOT_SUPPORTED	CLUSTER BY 列的类型为 BOOLEAN / ARRAY / MAP / STRUCT / BINARY。SDP 不会预先验证——提交前请使用 DESCRIBE 检查。聚类键必须是数值 / 字符串 / 日期 / 时间戳类型。完整类型规则请参阅 references/performance.md。
Cannot create streaming table from batch query	在流式表查询中，你写了 FROM read_files(...)（批处理）。请使用 FROM STREAM read_files(...)，以启动 Auto Loader。
摄取时出现 Column not found	schemaHints 与实际文件架构不匹配。请对示例文件执行 DESCRIBE，并调整提示。
流式读取因解析器错误而失败	文件摄取请使用 FROM STREAM read_files(...)，表间流式读取请使用 FROM stream(table)（或 FROM STREAM table_name——旧版 DLT 写法，建议使用函数形式）。不要混用。
Pipeline 在 serverless 上卡在 INITIALIZING	这是正常现象——首次运行需要几分钟进行冷启动。请勿终止。
Materialized View 未进行增量刷新	聚合的自动增量刷新需要 serverless，并且源表启用了 Delta 行跟踪（delta.enableRowTracking = true）。两者缺一不可，否则会退回到全量重新计算。用户询问增量刷新时，请说明 serverless 要求。
SCD2 查询无结果 / 在 START_AT 上提示“column not found”	Lakeflow 使用 __START_AT / __END_AT（双下划线）。当前行：WHERE __END_AT IS NULL。
事件输出中缺少 error.exceptions[0].message	你的 jq 读取的是 .message（其值只是“Update X is FAILED”）。请读取 error.exceptions[0].message 以获取真实原因——请参阅 2-rapid-iteration-with-cli.md。

发布模式

Pipeline 使用在其设置中配置的默认目录和架构。除非另行指定，所有数据集都会发布到该位置。

• 完全限定名称：在数据集名称中使用 catalog.schema.table，即可写入与 Pipeline 默认值不同的目录/架构。Pipeline 会直接在该位置创建数据集——无需 Sink。
• USE CATALOG / USE SCHEMA：可更改同一文件中后续所有定义的当前目录/架构的 SQL 命令。
• LIVE 前缀：已弃用。在默认发布模式下会被忽略。
• 在 Pipeline 中读取或定义数据集时，只使用数据集名称——除非 Pipeline 已采用完全限定名称，或用户明确要求使用不同的目标目录/架构，否则不要使用完全限定名称。

API 参考

为任何功能编写 Pipeline 代码之前，请先阅读链接的参考文件。 下表将各项功能映射到相应的 API，以及该功能和语言对应的详细说明文件。

某些功能基于其他功能构建——请同时阅读：

• Auto Loader / Auto CDC / Sinks 的目标是流式表 → 还需阅读 streaming-table-python.md / streaming-table-sql.md。
• Expectations 附加到数据集上 → 还需阅读数据集定义文件（streaming-table / materialized-view / temporary-view）。

数据集定义 API

功能	描述	Python	SQL	技能（Py）	技能（SQL）
流式表	持续增量处理，恰好一次，仅追加。	@dp.table() returning streaming DF	CREATE OR REFRESH STREAMING TABLE	streaming-table-python	streaming-table-sql
物化视图	物理存储查询结果，并进行增量刷新。	@dp.materialized_view()	CREATE OR REFRESH MATERIALIZED VIEW	materialized-view-python	materialized-view-sql
临时视图	仅限管道内部使用，不会持久化到 Unity Catalog。	@dp.temporary_view()	CREATE TEMPORARY VIEW	temporary-view-python	temporary-view-sql
持久化视图（UC）	发布到 UC；访问时运行查询（不存储数据）。	N/A — 仅 SQL	CREATE VIEW	—	view-sql
流式表（显式）	空目标表，由单独的流（追加流、AUTO CDC）填充。	dp.create_streaming_table()	CREATE OR REFRESH STREAMING TABLE（无 AS）	streaming-table-python	streaming-table-sql

Flow 和 Sink API

功能	描述	Python	SQL	技能（Py）	技能（SQL）
Append Flow	扇入：多个源 → 一个流式表。请使用此方式，而非 UNION。	@dp.append_flow()	CREATE FLOW ... INSERT INTO	streaming-table-python	streaming-table-sql
Backfill Flow	一次性加载历史数据，并将持续运行的实时流写入同一张表。	@dp.append_flow(once=True)	CREATE FLOW ... INSERT INTO ... ONCE	streaming-table-python	streaming-table-sql
Sink（Delta/Kafka/EH/自定义）	将流式输出写入外部 Delta / Kafka / Event Hubs。	dp.create_sink()	不适用 — 仅限 Python	sink-python	—
ForEachBatch Sink	自定义逐批 Python 逻辑（合并/更新插入、多目标）。公开预览。	@dp.foreach_batch_sink()	不适用 — 仅限 Python	foreach-batch-sink-python	—
RTM 更新 Flow	实时模式：将 Flow 路由到 Sink，实现亚秒级延迟。公开预览。	@dp.update_flow(target=...)	不适用 — 仅限 Python	real-time-mode	—

CDC API

功能	描述	Python	SQL	技能（Py）	技能（SQL）
Auto CDC（流式源）	从 CDC 数据流生成 SCD 类型 1（覆盖）或类型 2（历史记录）。	dp.create_auto_cdc_flow()	AUTO CDC INTO ... FROM STREAM	auto-cdc-python	auto-cdc-sql
Auto CDC（周期性快照）	比较连续的完整快照以检测变更。	dp.create_auto_cdc_from_snapshot_flow()	AUTO CDC ... FROM SNAPSHOT (DBR 18.3+)	auto-cdc-python	auto-cdc-sql

如需查询 SCD 类型 2 历史表（__START_AT / __END_AT、时间点查询、将事实表与历史维度表连接），请参阅 scd-2-querying.md。

数据质量 API

功能	描述	Python	SQL	技能（Python）	技能（SQL）
Expect（警告）	记录违规情况，保留所有行。	@dp.expect()	CONSTRAINT ... EXPECT (...)	expectations-python	expectations-sql
Expect 或丢弃	丢弃违规行。	@dp.expect_or_drop()	CONSTRAINT ... EXPECT (...) ON VIOLATION DROP ROW	expectations-python	expectations-sql
Expect 或失败	首次出现违规时使管道失败。	@dp.expect_or_fail()	CONSTRAINT ... EXPECT (...) ON VIOLATION FAIL UPDATE	expectations-python	expectations-sql
全部 Expect（警告）	一次检查多个约束，仅发出警告。	@dp.expect_all({})	多个 CONSTRAINT 子句	expectations-python	expectations-sql
全部 Expect 或丢弃	检查多个约束，出现违规时丢弃。	@dp.expect_all_or_drop({})	多个带 DROP ROW 的约束	expectations-python	expectations-sql
全部 Expect 或失败	检查多个约束，出现违规时失败。	@dp.expect_all_or_fail({})	多个带 FAIL UPDATE 的约束	expectations-python	expectations-sql

数据读取 API

功能	描述	Python	SQL	Skill (Py)	Skill (SQL)
批量读取（流水线数据集）	将同级表读取为静态 DataFrame。	spark.read.table("name")	SELECT ... FROM name	—	—
流式读取（流水线数据集）	将同级表读取为流式 DataFrame。	spark.readStream.table("name")	SELECT ... FROM STREAM(name)	—	—
Auto Loader（云文件）	从云存储增量摄取新文件。	spark.readStream.format("cloudFiles")	STREAM read_files(...)	auto-loader-python	auto-loader-sql
Kafka 源	从 Kafka 主题进行流式读取。	spark.readStream.format("kafka")	STREAM read_kafka(...)	kafka	kafka
Kinesis 源	从 AWS Kinesis 进行流式读取。	spark.readStream.format("kinesis")	STREAM read_kinesis(...)	—	—
Pub/Sub 源	从 GCP Pub/Sub 进行流式读取。	spark.readStream.format("pubsub")	STREAM read_pubsub(...)	—	—
Pulsar 源	从 Apache Pulsar 进行流式读取。	spark.readStream.format("pulsar")	STREAM read_pulsar(...)	—	—
Event Hubs 源	从 Azure Event Hubs（Kafka 协议）进行流式读取。	spark.readStream.format("kafka") + EH config	STREAM read_kafka(...) + EH config	kafka	kafka
JDBC / Lakehouse Federation	通过联合从外部系统批量读取数据。	spark.read.format("postgresql") 等	通过联合目录直接引用表	—	—
自定义数据源	用户定义的 Python 数据源。	spark.read[Stream].format("custom")	不适用 — 仅支持 Python	—	—
静态文件读取（批量）	一次性加载文件（不跟踪增量变化）。	spark.read.format("json"|"csv"|...).load()	read_files(...)（无 STREAM）	—	—
跳过上游变更提交	忽略上游表上的 CDC 提交。	.option("skipChangeCommits", "true")	STREAM name WITH (SKIPCHANGECOMMITS)	streaming-table-python	streaming-table-sql

表/架构功能 API

功能	描述	Python	SQL	技能（Py）	技能（SQL）
Liquid clustering	自适应的多列数据布局；取代 PARTITION + Z-ORDER。尽可能优先使用 Auto clustering	cluster_by=[...]	CLUSTER BY (col1, col2)	materialized-view-python	materialized-view-sql
Auto liquid clustering	Databricks 根据查询模式选择聚类键。	cluster_by_auto=True	CLUSTER BY AUTO	materialized-view-python	materialized-view-sql
分区列	旧版固定分区方式。优先使用 Liquid Clustering。	partition_cols=[...]	PARTITIONED BY (col1, col2)	materialized-view-python	materialized-view-sql
表属性	Delta 表属性（自动优化、CDF、保留期）。	table_properties={...}	TBLPROPERTIES (...)	materialized-view-python	materialized-view-sql
显式架构	预先声明列类型（而非推断得出）。	schema="col1 TYPE, ..."	(col1 TYPE, ...) AS	materialized-view-python	materialized-view-sql
生成列	写入时根据其他列计算得出的列。	schema="..., col TYPE GENERATED ALWAYS AS (expr)"	col TYPE GENERATED ALWAYS AS (expr)	materialized-view-python	materialized-view-sql
行筛选器（公开预览）	UC 精细化访问控制：通过函数筛选行。	row_filter="ROW FILTER fn ON (col)"	WITH ROW FILTER fn ON (col)	materialized-view-python	materialized-view-sql
列掩码（公开预览）	UC 精细化访问控制：通过函数对列进行掩码处理。	schema="..., col TYPE MASK fn USING COLUMNS (col2)"	col TYPE MASK fn USING COLUMNS (col2)	materialized-view-python	materialized-view-sql
私有数据集	已物化但未发布到 UC 的中间数据。	private=True	CREATE PRIVATE ...	materialized-view-python	materialized-view-sql

旧版 DLT 语法 — 始终迁移

上表仅展示现代 API。如果在用户代码中看到以下任何内容，这就是旧版 DLT 语法 — 始终迁移到现代形式，不要扩展旧版语法。建议修改前，请先阅读 references/dlt-migration.md，确保转换正确（尤其要注意 apply_changes → create_auto_cdc_flow 的语义，以及 partition_cols → cluster_by）。

如果看到……	……这就是 DLT。迁移为
import dlt	from pyspark import pipelines as dp
@dlt.table(...)、@dlt.append_flow(...)、@dlt.expect*	在 dp.* 上使用同名装饰器（例如，@dp.table、@dp.expect_or_drop）。
@dlt.view(...)（或旧版代码中出现的 @dp.view(...)）	@dp.temporary_view(...) — 现代 API 没有 view 装饰器，只有 temporary_view。
dlt.read("name") / dlt.read_stream("name")	spark.read.table("name") / spark.readStream.table("name")
dp.read(...) / dp.read_stream(...)	同样是旧版语法 — 使用 spark.read.table(...) / spark.readStream.table(...)。
dlt.apply_changes(...) / dp.apply_changes(...)	dp.create_auto_cdc_flow(...)。sequence_by 接受列名（字符串）或 col(...)；stored_as_scd_type 使用整数 2 表示类型 2，或字符串 "1" 表示类型 1。
dlt.apply_changes_from_snapshot(...)	dp.create_auto_cdc_from_snapshot_flow(...)
dlt.create_streaming_table(...)	dp.create_streaming_table(...)
SQL 中的 LIVE.<name> 前缀	裸名称（批处理使用 SELECT FROM name，流式处理使用 SELECT FROM STREAM(name)）。现代管道中使用 LIVE. 会报错。
CREATE LIVE TABLE / CREATE STREAMING LIVE TABLE	CREATE OR REFRESH MATERIALIZED VIEW / CREATE OR REFRESH STREAMING TABLE。
CREATE TEMPORARY LIVE VIEW（又称 CREATE LIVE VIEW）	CREATE TEMPORARY VIEW。例外：CREATE TEMPORARY VIEW 不支持用于期望的 CONSTRAINT 子句 — 对于少数需要在临时视图上设置期望的情况，仍保留 CREATE LIVE VIEW。请参阅 temporary-view-sql.md 和 expectations-sql.md。
APPLY CHANGES INTO ... FROM STREAM ...（SQL）	AUTO CDC INTO ... FROM STREAM ...
partition_cols=[...] / PARTITIONED BY (...) + ZORDER	cluster_by=[...] / CLUSTER BY (...)（Liquid Clustering）。
input_file_name()	_metadata.file_path（SQL）/ F.col("_metadata.file_path")（Python）。
create_streaming_table / 管道配置中的 target=... 参数	schema=...

语言选择（Python 或 SQL）

在搭建项目框架之前先做出选择——这会决定使用哪些模板文件（.py 或 .sql），以及适用哪些参考文档。两者可以共存，但要选定一个主要语言。不确定时，默认选择 SQL，因为它更简单。

用户信号	选择
“Python pipeline”、UDF、pandas、ML 推理、pyspark	Python
“SQL pipeline”、“SQL files”	SQL
“Simple pipeline”、“create a table”、“an aggregation”	SQL（更简单，默认使用）
复杂的参数化逻辑、自定义 UDF、ML	Python

如果含义不明确，请提问。除非用户明确要求切换，否则坚持使用选定的语言。

选择工作流

项目有三种形态——在搭建框架之前先选定一种。面向生产的工作默认选择 A，探索或演示项目默认选择 C。

• A：独立的新建 pipeline 项目（DAB）——pipeline 就是整个项目，不存在已有的 databricks.yml。使用 databricks pipelines init --output-dir . --config-file init-config.json 搭建框架。→ 1-project-initialization-with-dab.md
• B：在现有 bundle 中添加 pipeline（DAB）——databricks.yml 已存在。添加指向 src/ 的 resources/<name>.pipeline.yml。→ 1-project-initialization-with-dab.md#workflow-b-pipeline-in-existing-bundle
• C：快速 CLI 迭代（无 bundle）——用于原型开发。使用 databricks pipelines create / start-update / list-pipeline-events；如果项目要投入生产，再将其整理到 bundle 中。→ 2-rapid-iteration-with-cli.md

Pipeline 结构

• 除非用户另有要求，否则遵循奖牌架构模式（Bronze → Silver → Gold）。默认保持简单，只建少量表。
• 每个文件对应一个数据集，并以数据集命名。转换文件放在 src/ 或 transformations/ 中。
• Gold 层：保留关键业务维度。 聚合到 Gold 层时，应保留分析人员会用于筛选或切片的维度（位置、部门、产品线、客户细分、时间段）。过度聚合会丢失下游无法恢复的信息。如果提到了仪表板，其中的每个筛选条件都必须对应 Gold 表中的一列。在查询中进一步聚合很容易，但丢失的维度无法恢复。

运行 Pipeline

应根据上文选定的工作流来选择合适的运行命令。

• 工作流 A / B（DAB）——只有执行 databricks bundle deploy 后，代码更改才会生效。运行、试运行或选择性刷新前，务必先部署。

  databricks bundle validate --profile <profile>
  databricks bundle deploy -t dev --profile <profile>
  databricks bundle run <pipeline_name> -t dev --profile <profile>
  databricks pipelines get <pipeline_id> --profile <profile>      # status
  

  → DAB 运行和迭代的完整说明：references/1-project-initialization-with-dab.md#running-a-pipeline-workflow-a--b

• 工作流 C（CLI，无 bundle）——将文件上传到工作区，然后直接运行 pipeline。每次代码更改后都要重新上传。

  databricks workspace import-dir ./my_pipeline /Workspace/Users/<user>/my_pipeline --overwrite
  databricks pipelines start-update <pipeline_id>
  

  → CLI 运行和轮询方式的完整说明：references/2-rapid-iteration-with-cli.md

刷新模式（两种工作流均适用）：

• 选择性刷新适用于只需运行一张表的情况，优先使用。依赖项必须已物化。
• 完整刷新成本最高，也最危险，可能导致数据丢失（它会从头开始重新处理流式数据源，破坏流式状态）。只有确有必要时才使用。务必将其作为后续操作提出，并要求用户明确批准。

始终轮询更新，而不是顶层 pipeline 状态——轮询原因请参阅 2-rapid-iteration-with-cli.md#step-4-start-an-update-and-poll-that-update。bundle 运行也适用同一规则。

参考索引

项目与生命周期：

• 1-project-initialization-with-dab.md — 工作流 A 和 B。
• 2-rapid-iteration-with-cli.md — 工作流 C；启动更新 + 轮询 + 错误提取。
• pipeline-configuration.md — 完整的创建/更新 JSON 参考 + 变体片段 + 多架构 + 平台限制。
• performance.md — Liquid Clustering、状态管理、联接、预聚合、监控。
• dlt-migration.md — DLT → SDP 转换。

跨领域模式：

• streaming-patterns.md — 去重、窗口聚合、迟到数据、隔离救援数据、异常检测、延迟监控。
• scd-2-querying.md — 当前状态、特定时点查询、将事实表与历史维度表联接。
• kafka.md — Kafka / Event Hubs 数据摄取。
• real-time-mode.md — 实时模式（RTM）：亚秒级连续管道（@dp.update_flow、Kafka 接收器、无服务器或经典模式）。

Auto Loader 特定格式选项：JSON · CSV · XML · Parquet · Avro · Text · ORC。

数据集、流、CDC、预期、Auto Loader 和接收器参考按（功能、语言）列在上方的 API 参考表中。