📚 课程目录第4章数据摄取与ETL管道集成

4. 数据摄取与ETL管道集成

在 Databricks 机器学习平台上，模型效果的上限往往由数据质量决定，而数据质量的第一道关口就是摄取。数据摄取要解决的核心问题是：把外部数据库、文件系统、消息流中的数据，以可重复、可增量、可审计的方式落到 Delta Lake 表中，并让下游 Spark 与 Koalas（现为 pandas API on Spark）作业能够直接消费。本节围绕三条主线展开：批量与流式摄取方式的选择、Auto Loader 与 COPY INTO 的增量处理机制，以及摄取结果如何与 Spark、Koalas 集成。

一、摄取对象与方式选择

Databricks 的摄取源大致分为三类。第一类是数据库，如 MySQL、PostgreSQL、SQL Server、Oracle，通常通过 JDBC/ODBC 连接器读取，适合全量抽取或基于自增主键、更新时间戳的增量抽取。第二类是文件，包括对象存储（S3、ADLS、GCS）和云文件系统上的 CSV、JSON、Parquet、Avro、ORC，这类数据往往以“新文件到达”为增量信号。第三类是流，如 Kafka、Kinesis、Event Hubs，数据以持续追加的方式进入，需要流式处理语义。

选择摄取方式时，关键判断依据是数据到达模式与延迟要求。若数据以文件形式批量到达、且要求分钟级延迟，Auto Loader 是首选；若只是周期性执行一次性加载、且能接受手动触发，COPY INTO 更简单；若数据源是消息队列、要求秒级延迟，则应使用 Structured Streaming 直接读取。三者并非互斥，同一平台中常按数据源分别选用。

二、Auto Loader：面向文件到达的增量摄取

Auto Loader 通过 cloudFiles 数据源实现，核心能力是自动发现新到达的文件，并借助检查点与状态存储记录已处理文件，从而避免重复读取。它支持两种文件发现模式：目录列表模式（Directory Listing）和文件通知模式（File Notification）。前者通过定期列举目录发现新文件，配置简单，适合文件量中等的场景；后者依赖云存储的事件通知（如 S3 EventBridge、ADLS 队列），在文件量巨大时性能更好。

一个典型的 Auto Loader 流式读取配置如下：

df = (spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("cloudFiles.schemaLocation", "/mnt/checkpoints/schema")
      .option("cloudFiles.inferColumnTypes", "true")
      .load("/mnt/raw/events/"))

(df.writeStream
   .option("checkpointLocation", "/mnt/checkpoints/events")
   .trigger(availableNow=True)
   .toTable("bronze.events"))

这里 cloudFiles.schemaLocation 用于存放推断出的 Schema 及其演进历史；checkpointLocation 记录已处理文件与偏移量，是增量语义的关键。trigger(availableNow=True) 表示以批处理方式消费当前所有可用文件后停止，适合按调度周期运行的场景；若改为 processingTime='1 minute'，则持续微批处理。当新文件出现未预期字段时，Auto Loader 会将其放入 _rescued_data 列，而不是直接失败，这为 Schema 演进提供了缓冲。

三、COPY INTO：SQL 幂等加载

COPY INTO 是 Databricks SQL 层面的加载命令，适合从云存储向 Delta 表做幂等批量加载。它会记录已成功加载的文件，重复执行时自动跳过，因此可以安全地按计划反复运行。典型用法：

COPY INTO bronze.sales
FROM '/mnt/raw/sales/'
FILEFORMAT = CSV
FORMAT_OPTIONS ('header' = 'true', 'inferSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

与 Auto Loader 相比，COPY INTO 的优势是纯 SQL、无需维护流式作业，适合文件量不大、调度周期较长的场景；劣势是每次执行都要列举目录，文件数量极大时开销明显。选择时可参考：文件数量在百万级以上、要求低延迟，用 Auto Loader；文件数量在万级、按小时或按天调度，用 COPY INTO 即可。

四、与 Spark 和 Koalas 的集成

摄取落表后，下游消费统一通过 Delta 表进行。Spark 侧可直接用 spark.read.table("bronze.events") 或 SQL 查询，并利用 Delta 的时间旅行读取历史版本，便于回溯与重放。Koalas 侧则通过 databricks.koalas.read_delta 或 ks.read_table 读取同一张表，获得 pandas 风格的 API，同时底层仍由 Spark 分布式执行。

import databricks.koalas as ks

kdf = ks.read_table("bronze.events")
result = (kdf[kdf["event_type"] == "click"]
          .groupby("user_id")
          .size()
          .to_frame("click_cnt"))

需要特别注意的是，Koalas 的默认索引行为与 pandas 不完全一致，涉及 groupby、reset_index 时建议显式指定，避免分布式执行下索引语义偏差。此外，Koalas 适合中小规模特征加工，若数据量极大或需要复杂 join，直接使用 Spark DataFrame 往往更可控。

增量处理的关键在于“状态”而非“数据”本身：Auto Loader 用 schemaLocation 与 checkpointLocation 记录状态，COPY INTO 用表内元数据记录已加载文件。只要状态可恢复，作业中断后重跑就不会重复写入。

五、摄取管道的组织建议

实践中建议按 Bronze、Silver、Gold 分层组织。Bronze 层保留原始摄取结果，尽量少做转换，便于问题追溯；Silver 层做清洗、去重、类型规整；Gold 层面向特征与指标。摄取作业应统一使用 Delta 表作为落地目标，并开启 mergeSchema 以应对上游字段变化。对于流式源，建议在 Bronze 层先做 append，再在 Silver 层用 MERGE INTO 做去重与更新，这样既保证写入吞吐，又保证下游数据一致性。

最后，摄取管道必须可观测。应记录每次运行的输入文件数、写入行数、耗时与失败原因，并对 _rescued_data 非空比例设置告警。当该比例异常升高时，通常意味着上游 Schema 发生了未同步的变更，需要及时调整推断策略或显式声明 Schema。