SuperAPI AI 工具集
合作ENAI 行业资讯收藏
Skill 广场databricks-pipelines

databricks-pipelines

数据与分析

查看项目

更新于 10-15 08:55

技能介绍

用途

本技能用于在 Databricks 上开发 Lakeflow Spark 声明式流水线（原名 Delta Live Tables，相关缩写 SDP = LDP = Lakeflow Declarative Pipelines = DLT 可互换）。适用于构建批处理或流式数据管道，支持 Python 与 SQL 两种语言。

使用前须先用父技能 databricks-core 处理 CLI 基础、身份认证、profile 选择与数据发现命令。

主要能力

技能提供一个决策树，帮助确定数据集类型与所需特性。多个特性可叠加在同一数据集上（例如 Streaming Table 可同时使用 Auto Loader 摄入、Append Flows 扇入、Expectations 做数据质量）。应先选数据集类型，再叠加适用特性。

数据集类型：

  • 中间/可复用逻辑（不持久化）→ Temporary View；需发布到 Unity Catalog 供外部查询 → Persistent View（仅 SQL）。
  • 持久化数据集：源为周期性完整快照且需按 key 做 diff → Streaming Table + Auto CDC FROM SNAPSHOT（SQL 需 DBR 18.3+）；源为流式/增量/持续增长 → Streaming Table；源为批处理/历史/全量扫描 → Materialized View。
  • 输出到外部系统（仅 Python）→ Sink（Delta、Kafka、Event Hubs、自定义格式，或 Public Preview 的 ForEachBatch Sink）。
  • 数据质量约束 → Expectations（可附加于任意数据集类型）。

数据集定义 API：Streaming Table、Materialized View、Temporary View、Persistent View（仅 SQL）、显式 Streaming Table。

流与 Sink API：Append Flow（扇入，替代 UNION）、Backfill Flow（一次性历史加载 + 持续实时流）、Sink、ForEachBatch Sink（Public Preview）、RTM update flow（实时模式，Public Preview）。

CDC API：Auto CDC（流式源，SCD Type 1/2）、Auto CDC（周期快照）。查询 SCD Type 2 历史表使用 __START_AT / __END_AT。

数据质量 API：expect（警告）、expect_or_drop（丢弃违规行）、expect_or_fail（首次违规即失败），以及对应的批量约束版本 expect_all、expect_all_or_drop、expect_all_or_fail。

读取数据 API：批量读取、流式读取、Auto Loader、Kafka、Kinesis、Pub/Sub、Pulsar、Event Hubs、JDBC / Lakehouse Federation、自定义数据源、静态文件读取、跳过上游变更提交（skipChangeCommits）。

表/Schema 特性：Liquid clustering、Auto liquid clustering、分区列、表属性、显式 Schema、生成列、行过滤（Public Preview）、列掩码（Public Preview）、私有数据集。

使用方式

为任何特性编写流水线代码前，先阅读对应参考文件。参考表将每个特性映射到确切的 API 与 (特性, 语言) 的详情文件。部分特性叠加在其他特性之上，需同时阅读：如 Auto Loader / Auto CDC / Sink 指向 Streaming Table，需一并阅读 streaming-table-python.md / streaming-table-sql.md；Expectations 附加于数据集，需一并阅读对应数据集定义文件。

发布模式：流水线使用配置中设定的默认 catalog 和 schema，所有数据集默认发布到该处，除非被覆盖。可使用全限定名（catalog.schema.table）写到不同位置，无需 Sink；也可用 USE CATALOG / USE SCHEMA 改变后续定义的当前 catalog/schema。LIVE 前缀已弃用，在默认发布模式下被忽略。

常见陷阱

  • 未指定类型而要求“建表”→ 需询问源是流式还是批处理：流式源用 Streaming Table，批处理源用 Materialized View，配对错误会在验证时报错。
  • 对流式源做聚合 → 使用 Materialized View 做批量读取（spark.read.table / 不带 STREAM 的 SELECT FROM）。
  • 中间逻辑 → 默认使用 Temporary View。
  • 流合并 → 使用多个 Append Flows，跨流式源 UNION 是反模式。
  • 变更数据集类型 → 不能就地 ST↔MV 切换，全量刷新无效，需手动删除或重命名。
  • 多列 Auto CDC 排序 → SQL 用 SEQUENCE BY STRUCT(col1, col2)；Python 用 sequence_by=struct("col1", "col2")。
  • Python 独有特性 → Sink、ForEachBatch Sink、自定义数据源仅 Python 支持。
  • 应推荐一种清晰方案，不要罗列反模式或次优替代。

常见问题排查

技能列出常见错误与成因/修复映射，例如：拒绝 CREATE OR REPLACE STREAMING TABLE（应使用 CREATE OR REFRESH）；CLI 操作 /Volumes/... 需加 dbfs: 前缀；DELTA_CLUSTERING_COLUMNS_DATATYPE_NOT_SUPPORTED 因 CLUSTER BY 列类型不支持；从批查询创建流式表需用 FROM STREAM read_files(...)；schemaHints 与实际文件 Schema 不匹配导致列找不到；流式读取解析错误需区分文件摄入与表到表的写法；serverless 流水线长时间处于 INITIALIZING 属正常冷启动；Materialized View 增量刷新需 serverless + Delta 行跟踪（delta.enableRowTracking = true）；SCD2 查询使用双下划线 __START_AT / __END_AT；读取错误真正原因应取 error.exceptions[0].message 而非 .message。

旧版 DLT 语法迁移

若用户代码中出现 import dlt、@dlt.table(...)、@dlt.view(...)、apply_changes、partition_cols 等旧版 DLT 语法，应迁移到现代 API（如 from pyspark import pipelines as dp、@dp.table、create_auto_cdc_flow、cluster_by）。建议改动前先阅读 references/dlt-migration.md 以确保转换正确（尤其是 apply_changes → create_auto_cdc_flow 的语义与 partition_cols → cluster_by）。

适用场景与限制

适用场景为在 Databricks 上开发批处理或流式数据管道，覆盖从摄入（Auto Loader、消息总线）、转换（Streaming Table、Materialized View）、变更捕获（Auto CDC）到输出（Sink）及数据质量（Expectations）的完整流程，并支持 Python 与 SQL。

限制：部分特性为版本或预览要求，例如 Auto CDC FROM SNAPSHOT 的 SQL 需 DBR 18.3+，ForEachBatch Sink 与 RTM 为 Public Preview，行过滤与列掩码为 Public Preview，Sink/ForEachBatch/自定义数据源仅 Python。Materialized View 的自动增量刷新需 serverless 与 Delta 行跟踪同时满足。