📚 课程目录第4章数据摄取管道构建

4. 数据摄取管道构建

在 Databricks 湖仓平台中，数据摄取是所有下游 ETL、特征工程与 BI 分析的第一道关口。本章围绕四条主线展开：Auto Loader 的增量文件发现机制、COPY INTO 的幂等批量加载、流式与批量摄取的选型边界，以及Schema 推断与演化、错误处理与死信队列的工程化落地。最后通过一个多源异构接入实战，把对象存储、消息队列与关系库三类数据源统一到 Bronze 层。

一、Auto Loader：增量文件发现与 Schema 演化

Auto Loader 通过 cloudFiles 数据源实现，核心是两种文件发现模式：目录列表模式（Directory Listing）与文件通知模式（File Notification）。前者周期性列举输入目录，适合文件量在百万级以下、目录结构简单的场景；后者依赖云存储事件（如 S3 EventBridge、ADLS Event Grid、GCS Pub/Sub）写入通知队列，Databricks 消费队列获得新文件路径，可支撑数十亿文件的低延迟发现。

Schema 推断由 cloudFiles.schemaLocation 持久化，首次运行采样推断并写入 _schemas 目录。当新文件出现新列时，通过 cloudFiles.schemaEvolutionMode 控制行为：addNewColumns（默认，自动加列并重启流）、rescue（新列进入 _rescued_data）、failOnNewColumns（报错停止）、none（忽略）。类型冲突（如 string 变 int）不会自动升级，会落入 rescued 列，需人工介入。

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  .option("cloudFiles.schemaLocation", "abfss://meta@acct.dfs.core.windows.net/schema/orders") \
  .option("cloudFiles.schemaEvolutionMode", "addNewColumns") \
  .option("cloudFiles.useNotifications", "true") \
  .option("cloudFiles.includeExistingFiles", "true") \
  .load("abfss://raw@acct.dfs.core.windows.net/orders/") \
  .writeStream.option("checkpointLocation", "abfss://meta@acct.dfs.core.windows.net/ckpt/orders") \
  .trigger(availableNow=True) \
  .toTable("bronze.orders")

其中 trigger(availableNow=True) 是推荐的“准流式”批处理方式：一次性消费所有可用文件后停止，兼顾增量语义与成本控制。

二、COPY INTO：幂等批量加载

COPY INTO 是 SQL 层的幂等加载命令，适合按天/小时调度的批量场景。它通过记录已加载文件避免重复，支持 FILES 指定、PATTERN 通配、FORMAT_OPTIONS 与 COPY_OPTIONS。与 Auto Loader 的关键差异在于：COPY INTO 是批处理语义，不维护长驻流，Schema 演化能力弱，遇到新列需显式 MERGE SCHEMA。

COPY INTO bronze.events
FROM 'abfss://raw@acct.dfs.core.windows.net/events/2024/06/'
FILEFORMAT = PARQUET
PATTERN = '*.parquet'
COPY_OPTIONS ('mergeSchema' = 'true', 'force' = 'false');

选型原则：文件持续到达、需要低延迟与 Schema 演化 → Auto Loader；周期性大批量、文件已就绪、追求 SQL 简洁 → COPY INTO。两者可共存，例如历史回填用 COPY INTO，增量用 Auto Loader。

三、错误处理与死信队列

摄取管道必须区分三类异常：可解析但业务非法（如金额为负）、格式损坏（JSON 截断、Parquet 校验失败）、Schema 冲突（类型不匹配）。Auto Loader 通过 _rescued_data 列捕获解析失败与新增字段；对于业务校验失败，应在 Bronze→Silver 之间用 expectations 或显式过滤写入死信表。

-- 死信表结构示例
CREATE TABLE IF NOT EXISTS bronze.orders_dlq (
  raw_payload STRING,
  source_file STRING,
  ingest_ts TIMESTAMP,
  error_type STRING,
  error_detail STRING
);

推荐做法：在流中先用 _rescued_data IS NOT NULL 分流，将坏记录连同源文件路径写入 DLQ，正常记录继续入 Bronze。DLQ 表按天分区，便于回溯与重放。

四、多源异构接入实战

假设需要接入三类源：对象存储中的 JSON 订单、Kafka 中的点击流、MySQL 中的用户维表。统一落到 Bronze 层三张表，再在 Silver 层做清洗与关联。下图展示了三条摄取链路的对象、协议、关键参数与落表关系。

三条链路的落地要点：订单链路启用通知模式与 addNewColumns，checkpoint 与 schema 目录分离存放；点击流链路设置 maxOffsetsPerTrigger 控制单批吞吐，用 watermark 处理迟到数据；用户维表链路先用 JDBC 全量拉取到临时视图，再用 MERGE INTO 按 user_id 做 upsert，避免重复。所有链路统一写入 _ingest_ts 与 _source_file 元数据列，便于血缘追踪。

建议在 Bronze 层保留原始字段与 rescued 列，不做强类型转换；类型清洗与业务校验放到 Silver 层，这样 DLQ 重放时无需回源。

通知模式需要额外配置云存储事件与队列权限，若权限未就绪，Auto Loader 会回退到目录列表模式，表现为延迟升高但不会报错，需通过 cloudFiles.useNotifications 与队列指标确认实际模式。

综上，摄取管道的工程化核心在于：选对发现机制、管好 Schema 演化、隔离坏数据、统一元数据。这四点决定了 Bronze 层能否稳定支撑后续所有下游消费。