在 Databricks 上开发 Lakeflow…

用 Python 或 SQL 在 Databricks 上构建批处理或流式数据管道（Lakeflow/DLT/SDP）时，按决策树选数据集类型、避开常见陷阱并正确部署运行。

适用场景

解决思路

1. 先使用父技能 databricks-core 处理 CLI 基础、认证、profile 选择和数据发现命令。
2. 按决策树先确定数据集类型，再叠加适用特性：中间/可复用逻辑（不持久化）→ Temporary View；需要发布到 UC 供外部查询 → Persistent View（仅 SQL）；持久化数据集且源是流式/增量/持续增长 → Streaming Table；源是批处理/历史/全量扫描 → Materialized View；输出到外部系统（仅 Python）→ Sink；数据质量约束 → Expectations（可加在任何数据集类型上）。
3. 流式表的具体分支：文件摄取（云存储、Volumes）用 Auto Loader；消息总线（Kafka、Kinesis、Pub/Sub、Pulsar、Event Hubs）用流式源读取；已有流式/Delta 表用流式读取；CDC/upsert/跟踪变更/按键保留最新/SCD Type 1 或 2 用 Auto CDC；多源汇入一张表用 Append Flows（不要用 union）；历史回填+实时流用一次性 Append Flow + 常规 flow；带水印的窗口聚合用有状态流处理。
4. 物化视图的具体分支：全数据集聚合/join（GROUP BY、SUM、COUNT 等）；从流式表做 Gold 层聚合时用批读取（spark.read / 不带 STREAM）；JDBC/Federation/外部批源；小型静态文件加载（参考数据，不做流式读取）。
5. Sink 的具体分支：已有外部表且不由本管道管理 → format="delta" 的 Sink（若管道应拥有该表，优先用全限定数据集名）；Kafka/Event Hubs → format="kafka" 的 Sink + @dp.append_flow(target="sink_name")；原生不支持的定制目标 → 自定义 format 的 Sink；每批需要自定义 merge/upsert 逻辑 → ForEachBatch Sink（Public Preview）；每批多个目标 → ForEachBatch Sink（Public Preview）。
6. 避开常见陷阱：SDP = LDP = Lakeflow Declarative Pipelines =（原）DLT，用户提到时都视为同一事物；用户说“创建一张表”但未指定类型时，先问源是流式还是批处理（流式→Streaming Table，批处理→Materialized View，配错会在校验时报错）；对流式源做聚合要用 Materialized View 加批读取（spark.read.table / 不带 STREAM 的 SELECT FROM），因为 ST 只追加、源行变化时不会重算聚合，而 MV 会；中间逻辑默认用 Temporary View，即使是被多个下游表共享的逻辑也一样，只有物化一次能显著节省重处理时才用 Private MV/ST（private=True / CREATE PRIVATE ...），Auto CDC 前的预处理必须用 temp view（CDC flow 从 STREAM(view_name)（SQL）或 spark.readStream.table("view_name")（Python）读取）。
7. 更多陷阱：流的 union 要用多个 Append Flows，跨流式源 UNION 是反模式；不能在原地把 ST 改成 MV 或 MV 改成 ST，full refresh 也没用，必须手动 drop 已有表或重命名新数据集；SQL 数据集 CREATE OR REFRESH 和 CREATE 都能解析，但 CREATE OR REFRESH 是惯用写法，PRIVATE 数据集用 CREATE OR REFRESH PRIVATE STREAMING TABLE / ... MATERIALIZED VIEW；Kafka/Event Hubs sink 序列化时 value 列是必需的，用 to_json(struct(*)) AS value 序列化行；多列 Auto CDC 排序：SQL 用 SEQUENCE BY STRUCT(col1, col2)，Python 用 sequence_by=struct("col1", "col2")；Auto CDC TRUNCATE（仅 SCD Type 1）：SQL 用 APPLY AS TRUNCATE WHEN condition，Python 用 apply_as_truncates=expr("condition")，不要声称不支持 truncate；仅 Python 的特性：Sinks、ForEachBatch Sinks、从快照做 CDC、自定义数据源，用户用 SQL 时要说明并建议切换到 Python；只推荐一种清晰方案，不要罗列反模式或劣质替代方案，只在确实有不同权衡时才提替代方案。
8. 按常见错误对照表修复：CREATE OR REPLACE STREAMING TABLE / MATERIALIZED VIEW 被拒绝 → 这是标准 SQL 不是 SDP，改用 CREATE OR REFRESH STREAMING TABLE / CREATE OR REFRESH MATERIALIZED VIEW；databricks fs ls /Volumes/... 报 CLI 错误 → 即使是 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 确认，聚簇键必须是数值/字符串/日期/时间戳；报 Cannot create streaming table from batch query → 流式表查询里写了 FROM read_files(...)（批处理），改用 FROM STREAM read_files(...) 以启用 Auto Loader；摄取时报 Column not found → schemaHints 与实际文件 schema 不匹配，DESCRIBE 一个样本文件并对齐 hints；流式读取报解析错误 → 文件摄取用 FROM STREAM read_files(...)，表到表流用 FROM stream(table)（或 FROM STREAM table_name，旧版 DLT，优先函数形式），不要混用；serverless 下管道卡在 INITIALIZING → 正常，首次运行冷启动需要几分钟，不要杀掉；Materialized View 不增量刷新 → 聚合的自动增量刷新需要 serverless + 源上启用 Delta row tracking（delta.enableRowTracking = true），两者缺一就退化为全量重算，用户问增量刷新时要说明 serverless 要求；SCD2 查询返回空或 START_AT 报 column not found → Lakeflow 用 __START_AT / __END_AT（双下划线），当前行用 WHERE __END_AT IS NULL；events 输出里缺 error.exceptions[0].message → 你的 jq 读的是 .message（只是 "Update X is FAILED"），要读 error.exceptions[0].message 才能拿到真正原因。
9. 发布模式：管道在设置里配置默认 catalog 和 schema，所有数据集默认发布到那里，除非被覆盖；用 catalog.schema.table 全限定名可写到与管道默认不同的 catalog/schema，管道会直接在那里创建数据集，不需要 Sink；USE CATALOG / USE SCHEMA 是 SQL 命令，改变同一文件中后续所有定义的当前 catalog/schema；LIVE 前缀已废弃，在默认发布模式下被忽略；在管道内读取或定义数据集时只用数据集名，不要用全限定名，除非管道本身已这样做或用户明确要求写到不同目标 catalog/schema。
10. 写任何特性的管道代码前先读对应参考文件；有些特性叠加在其他特性之上要同时读：Auto Loader / Auto CDC / Sinks 目标是流式表 → 同时读 streaming-table-python.md / streaming-table-sql.md；Expectations 附着在数据集上 → 同时读该数据集定义文件（streaming-table / materialized-view / temporary-view）。
11. 数据集定义 API：Streaming Table（Python @dp.table() 返回流式 DF，SQL CREATE OR REFRESH STREAMING TABLE）；Materialized View（@dp.materialized_view()，CREATE OR REFRESH MATERIALIZED VIEW）；Temporary View（@dp.temporary_view()，CREATE TEMPORARY VIEW）；Persistent View（仅 SQL，CREATE VIEW）；Streaming Table 显式形式（dp.create_streaming_table()，CREATE OR REFRESH STREAMING TABLE 不带 AS）。
12. Flow 与 Sink API：Append Flow（@dp.append_flow()，CREATE FLOW ... INSERT INTO）；Backfill Flow（@dp.append_flow(once=True)，CREATE FLOW ... INSERT INTO ... ONCE）；Sink（dp.create_sink()，仅 Python）；ForEachBatch Sink（@dp.foreach_batch_sink()，仅 Python，Public Preview）；RTM update flow（@dp.update_flow(target=...)，仅 Python，Public Preview）。
13. CDC API：Auto CDC（流式源，dp.create_auto_cdc_flow()，AUTO CDC INTO ... FROM STREAM）；Auto CDC（周期快照，dp.create_auto_cdc_from_snapshot_flow()，仅 Python）；查询 SCD Type 2 历史表（__START_AT / __END_AT、时间点、事实与历史维度 join）见 scd-2-querying.md。
14. 数据质量 API：Expect（warn，@dp.expect()，CONSTRAINT ... EXPECT (...)）；Expect or drop（@dp.expect_or_drop()，CONSTRAINT ... EXPECT (...) ON VIOLATION DROP ROW）；Expect or fail（@dp.expect_or_fail()，CONSTRAINT ... EXPECT (...) ON VIOLATION FAIL UPDATE）；Expect all（warn，@dp.expect_all({})，多个 CONSTRAINT 子句）；Expect all or drop（@dp.expect_all_or_drop({})，多个带 DROP ROW 的约束）；Expect all or fail（@dp.expect_all_or_fail({})，多个带 FAIL UPDATE 的约束）。
15. 读取数据 API：批读取管道数据集（spark.read.table("name")，SELECT ... FROM name）；流式读取管道数据集（spark.readStream.table("name")，SELECT ... FROM STREAM(name)）；Auto Loader（spark.readStream.format("cloudFiles")，STREAM read_files(...)）；Kafka 源（spark.readStream.format("kafka")，STREAM read_kafka(...)）；Kinesis 源（spark.readStream.format("kinesis")，STREAM read_kinesis(...)）；Pub/Sub 源（spark.readStream.format("pubsub")，STREAM read_pubsub(...)）；Pulsar 源（spark.readStream.format("pulsar")，STREAM read_pulsar(...)）；Event Hubs 源（spark.readStream.format("kafka") + EH 配置，STREAM read_kafka(...) + EH 配置）；JDBC / Lakehouse Federation（spark.read.format("postgresql") 等，通过 federation catalog 直接引用表）；自定义数据源（spark.read[Stream].format("custom")，仅 Python）；静态文件批读取（spark.read.format("json"|"csv"|...).load()，read_files(...) 不带 STREAM）；跳过上游变更提交（.option("skipChangeCommits", "true")，STREAM name WITH (SKIPCHANGECOMMITS)）。
16. 表/Schema 特性 API：Liquid clustering（cluster_by=[...]，CLUSTER BY (col1, col2)，尽量优先用 Auto clustering）；Auto liquid clustering（cluster_by_auto=True，CLUSTER BY AUTO）；Partition columns（partition_cols=[...]，PARTITIONED BY (col1, col2)，优先用 Liquid Clustering）；Table properties（table_properties={...}，TBLPROPERTIES (...)）；Explicit schema（schema="col1 TYPE, ..."，(col1 TYPE, ...) AS）；Generated columns（schema="..., col TYPE GENERATED ALWAYS AS (expr)"，col TYPE GENERATED ALWAYS AS (expr)）；Row filter（Public Preview，row_filter="ROW FILTER fn ON (col)"，WITH ROW FILTER fn ON (col)）；Column mask（Public Preview，schema="..., col TYPE MASK fn USING COLUMNS (col2)"，col TYPE MASK fn USING COLUMNS (col2)）；Private dataset（private=True，CREATE PRIVATE ...）。
17. 遇到旧版 DLT 语法一律迁移，不要在其上扩展，改前先读 dlt-migration.md：import dlt → from pyspark import pipelines as dp；@dlt.table(...)、@dlt.append_flow(...)、@dlt.expect* → dp.* 上同名装饰器；@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 表示 Type 2、字符串 "1" 表示 Type 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 子句做 expectations，极少数需要在 temp view 上加 expectations 时保留 CREATE LIVE VIEW；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=...。
18. 先决定语言（Python vs SQL）再搭脚手架，因为它决定模板文件（.py vs .sql）和适用参考文档；两者可共存但选一个主语言；不确定时默认 SQL 更简单。信号："Python pipeline"、UDF、pandas、ML 推理、pyspark → Python；"SQL pipeline"、"SQL files" → SQL；"简单管道"、"创建一张表"、"一个聚合" → SQL（默认）；复杂参数化逻辑、自定义 UDF、ML → Python；有歧义就问，除非用户明确切换否则坚持已选语言。
19. 先选项目形态再搭脚手架，生产用途默认 A，探索/演示脚手架默认 C：A 独立新管道项目（DAB）——管道就是项目，没有现成 databricks.yml，用 databricks pipelines init --output-dir . --config-file init-config.json 搭脚手架；B 已有 bundle 中的管道（DAB）——databricks.yml 已存在，添加指向 src/ 的 resources/<name>.pipeline.yml；C 快速 CLI 迭代（无 bundle）——原型阶段，用 databricks pipelines create / start-update / list-pipeline-events，之后若要上生产再形式化为 bundle。
20. 管道结构：除非用户另有说明，遵循 medallion 模式（Bronze → Silver → Gold），默认保持简单，只建几张表；一个数据集一个文件，以数据集命名，转换文件放在 src/ 或 transformations/；Gold 层保留关键业务维度——聚合到 Gold 时保留分析师会用来过滤/切片的维度（地点、部门、产品线、客户细分、时间段），过度聚合会丢失下游无法恢复的信息，若提到仪表盘则其每个过滤器都要是 Gold 表的列，在查询里进一步聚合比恢复丢失维度更容易。
21. 运行管道：Workflow A/B（DAB）——代码改动只有 databricks bundle deploy 后才生效，任何 run、dry run 或 selective refresh 前都要先 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> 查状态。Workflow C（CLI，无 bundle）——把文件上传到 workspace 后直接驱动管道，每次改代码后重新上传：databricks workspace import-dir ./my_pipeline /Workspace/Users/<user>/my_pipeline --overwrite；databricks pipelines start-update <pipeline_id>。
22. 刷新模式（两种 workflow 通用）：只跑一张表时优先用 selective refresh，依赖必须已物化；full refresh 最昂贵也最危险，可能导致数据丢失（会从头重处理流式源、销毁流式状态），只在确实必要时用，且始终作为需要用户明确批准的后续建议提出。
23. 始终轮询 update 而不是顶层管道状态（bundle run 同理）。
24. 参考索引：项目与生命周期——1-project-initialization-with-dab.md（Workflow A 和 B）、2-rapid-iteration-with-cli.md（Workflow C；start-update + 轮询 + 错误提取）、pipeline-configuration.md（完整 create/update JSON 参考 + 变体片段 + 多 schema + 平台约束）、performance.md（Liquid Clustering、状态管理、join、预聚合、监控）、dlt-migration.md（DLT → SDP 转换）；横切模式——streaming-patterns.md（去重、窗口聚合、迟到数据、rescue-data 隔离、异常检测、延迟监控）、scd-2-querying.md（当前状态、时间点、事实与历史维度 join）、kafka.md（Kafka / Event Hubs 摄取）、real-time-mode.md（RTM：亚秒级连续管道，@dp.update_flow、Kafka sinks、serverless 或 classic）；Auto Loader 各格式选项：JSON、CSV、XML、Parquet、Avro、Text、ORC。

适用边界 · 注意事项

• 没有安装 databricks CLI（>= v1.0.0）时无法使用本技能。
• 不涉及 Databricks 平台或 Lakeflow/DLT/SDP 管道的通用数据工程任务不适用。
• 需要付费 Databricks 账号与 Unity Catalog 环境，纯本地或非 Databricks 环境不适用。