4. 数据摄取基础

在 Databricks 批流一体架构中，数据摄取是整条数据链路的入口。无论下游是做实时看板、离线报表还是特征工程，第一步都要把外部系统的数据可靠地引入 Delta Lake。Databricks 主要面对三类数据源：云对象存储（S3、ADLS Gen2、GCS）中的文件、关系型数据库（MySQL、PostgreSQL、SQL Server 等）中的表，以及消息队列（Kafka、Event Hubs、Kinesis）中的事件流。这三类源的读取语义差异很大：文件是"批量到达、可重复扫描"的，数据库是"有主键、可增量拉取"的，消息队列是"持续追加、按偏移量消费"的。理解这些差异，才能选对摄取工具。

三类数据源的读取方式与工具选择

云存储文件是最常见的摄取对象。Databricks 提供两条路径：COPY INTO 和 Auto Loader。COPY INTO 是一条 SQL 命令，适合按计划（如每小时）批量加载新文件，它通过记录已处理文件列表实现幂等，重复执行不会重复加载。Auto Loader 则基于 Spark Structured Streaming，使用 cloudFiles 数据源，能够以流式方式持续发现新文件，并借助 RocksDB 状态存储跟踪已见文件，支持目录列表和文件通知两种发现模式。

关系型数据库通常通过 JDBC 读取。批量场景可用 spark.read.jdbc() 配合分区列并行拉取；增量场景则依赖时间戳列或自增主键做水位线（watermark）过滤，例如每次只拉取 updated_at > 上次最大值 的行。消息队列使用 Kafka 连接器，以流式方式消费，通过 startingOffsets 和 checkpointLocation 保证断点续读。

下面的图展示了三类源进入 Delta Lake 的摄取路径与关键机制，注意 Auto Loader 与 COPY INTO 在"增量识别"上的不同实现。

Auto Loader 增量加载实操

Auto Loader 的核心是 cloudFiles 数据源。下面示例从 ADLS Gen2 的 raw/events 目录持续读取 JSON 文件，写入 Bronze 层 Delta 表，并开启 schema 自动演进。

checkpoint = "abfss://meta@acct.dfs.core.windows.net/ckpt/events"

df = (spark.readStream
      .format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("cloudFiles.schemaLocation",
              "abfss://meta@acct.dfs.core.windows.net/schema/events")
      .option("cloudFiles.inferColumnTypes", "true")
      .load("abfss://raw@acct.dfs.core.windows.net/events/"))

(df.writeStream
   .option("checkpointLocation", checkpoint)
   .option("mergeSchema", "true")
   .trigger(availableNow=True)
   .toTable("bronze.events"))

关键参数说明：cloudFiles.schemaLocation 保存推断出的 schema 及演进历史；checkpointLocation 记录已处理文件与流进度；trigger(availableNow=True) 表示"处理完当前所有新文件后停止"，适合按调度运行的准实时场景，而 trigger(processingTime="1 minute") 则用于常驻流。若目录文件量极大，可启用文件通知模式：.option("cloudFiles.useNotifications", "true")，由云存储事件驱动发现，避免全目录列举的开销。

COPY INTO 增量加载实操

COPY INTO 更适合"定时批量、文件量中等"的场景。它自动跳过已加载文件，语法简洁：

COPY INTO bronze.events
FROM 'abfss://raw@acct.dfs.core.windows.net/events/'
FILEFORMAT = JSON
FORMAT_OPTIONS ('mergeSchema' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

执行后可通过 DESCRIBE HISTORY bronze.events 查看每次加载的 operationMetrics，其中 numFiles、numOutputRows 反映本次实际新增的文件数与行数。若某次执行 numFiles=0，说明没有新文件，属于正常幂等行为。需要重新加载时，可用 COPY INTO ... FORCE = true 强制重读，但要注意这会造成重复数据，通常只在修复错误后配合去重使用。

数据库与消息队列的增量读取

数据库增量拉取依赖水位线。假设源表有 updated_at 列，首次全量后记录最大时间戳，后续每次执行：

last_ts = spark.sql("SELECT max(updated_at) FROM bronze.orders").collect()[0][0]

jdbc_df = (spark.read.format("jdbc")
  .option("url", "jdbc:postgresql://db:5432/shop")
  .option("dbtable", f"(SELECT * FROM orders WHERE updated_at > '{last_ts}') t")
  .option("user", "etl").option("password", "***")
  .option("partitionColumn", "order_id")
  .option("lowerBound", 1).option("upperBound", 1000000)
  .option("numPartitions", 8)
  .load())

消息队列则用 Kafka 连接器，startingOffsets 设为 earliest 或 latest，checkpoint 保证重启后从上次 offset 继续。Kafka 消息的 value 通常是二进制，需 CAST(value AS STRING) 后再解析 JSON。