📚 课程目录第4章批处理数据摄取实战

4. 批处理数据摄取实战

批处理数据摄取是数据湖仓的第一道关口。它的目标不是“把文件拷进来”这么简单，而是要把云存储、关系型数据库、跨组织共享数据这三类异构来源，统一收敛成可治理、可增量、可重放的 Bronze 表。Bronze 层保留原始语义、附加摄取元数据（来源、时间、文件路径、批次号），为后续 Silver 清洗和 Gold 聚合提供可信底座。本节围绕三条摄取路径展开：云存储批量加载、JDBC 批量读取、Delta Sharing 共享读取，并重点讲解 Auto Loader 如何用增量发现机制替代全量扫描。

一、三条摄取路径的定位与选择

云存储适合落地原始文件（JSON、CSV、Parquet、Avro），数据量大、schema 可能漂移，是 Auto Loader 的主战场。JDBC 适合从 MySQL、PostgreSQL、SQL Server、Oracle 等业务库拉取结构化表，通常按主键或时间戳做增量切片。Delta Sharing 适合跨组织、跨工作区共享 Delta 表，接收方无需复制底层存储，直接以只读方式读取共享表。三者的共同终点都是 Bronze 表，但读取语义、增量方式和失败重试策略不同。

二、云存储批量加载与 Auto Loader 增量发现

传统 spark.read.json("s3://raw/orders/") 每次都会全量列出目录，文件一多就慢，而且无法感知“哪些文件是新的”。Auto Loader 通过 cloudFiles 数据源解决这个问题：它维护一个 checkpoint 记录已处理文件，并可选维护 schemaLocation 记录推断出的 schema。新文件到达时只处理增量部分，这就是“增量发现”。

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "s3://meta/orders/schema")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
  .load("s3://raw/orders/"))

(df.writeStream
  .option("checkpointLocation", "s3://meta/orders/checkpoint")
  .trigger(availableNow=True)
  .toTable("bronze.orders"))

关键参数说明：cloudFiles.format 指定源文件格式；schemaLocation 存放推断 schema，避免每次重启重新推断；schemaEvolutionMode 设为 addNewColumns 时，新出现的字段会自动加入 Bronze 表，旧字段保持不变；trigger(availableNow=True) 表示“把当前所有新文件处理完就停”，这正是批处理场景下用流式引擎做增量批加载的典型写法。checkpoint 目录必须唯一且稳定，否则会重复消费或漏读。

Auto Loader 的“增量”依赖 checkpoint 中记录的文件清单。如果手动删除 checkpoint，下一次运行会把所有历史文件当作新文件重新处理，导致 Bronze 重复。生产环境应把 checkpoint 与 schemaLocation 放在独立、受版本控制或受生命周期策略保护的路径下。

三、JDBC 批量读取与并行切片

JDBC 读取的核心是“怎么把一张大表切成多个分区并行拉取”。Spark 提供四个参数：partitionColumn、lowerBound、upperBound、numPartitions。以 orders 表按 id 切片为例，假设 id 范围 1 到 1,000,000，设 numPartitions=8，则每个分区约 125,000 行，8 个任务并行读取。

jdbc_df = (spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql://db:5432/sales")
  .option("dbtable", "public.orders")
  .option("user", "etl_reader")
  .option("password", dbutils.secrets.get("db", "etl_pwd"))
  .option("partitionColumn", "id")
  .option("lowerBound", "1")
  .option("upperBound", "1000000")
  .option("numPartitions", "8")
  .load())

jdbc_df.write.mode("append").saveAsTable("bronze.orders_jdbc")

增量切片常用两种策略：一是按自增主键记录上次最大 id，下次只读 id > last_max_id；二是按 updated_at 时间戳，读取 updated_at > last_watermark。前者适合只追加的表，后者适合有更新的表。注意 partitionColumn 必须是数值、日期或时间戳类型，且分布尽量均匀，否则会出现数据倾斜。

四、Delta Sharing 批量读取

Delta Sharing 的读取方只需配置 profile 文件，即可用 deltaSharing 格式读取共享表。它天然支持按版本读取，便于做可重放的批处理。

share_df = (spark.read
  .format("deltaSharing")
  .option("profile", "/etc/databricks/secrets/share_profile.share")
  .option("responseFormat", "parquet")
  .table("sales_share.orders"))

share_df.write.mode("overwrite").saveAsTable("bronze.orders_shared")

若共享方持续更新，可用 versionAsOf 或 timestampAsOf 读取指定版本，实现幂等重跑。Delta Sharing 是只读的，接收方不能写回共享表，因此 Bronze 落地是必要的中间步骤。

五、落地 Bronze 的通用规范

三条路径最终都写入 Delta 表，建议统一附加元数据列：_ingest_time（摄取时间）、_source_file（源文件路径或表名）、_batch_id（批次标识）。写入模式上，云存储和 Delta Sharing 常用 append 或 overwrite，JDBC 增量常用 append。Bronze 表应保留原始字段类型，不做业务清洗，只做必要的类型统一和元数据补充，确保下游可以按批次重放。

来源	读取方式	增量机制	典型写入模式
云存储	cloudFiles	checkpoint 文件清单	append
JDBC	spark.read.jdbc	主键 / 时间戳水位	append
Delta Sharing	deltaSharing	版本号 / 时间戳	overwrite 或 append

Auto Loader 的 schema 推断在首次遇到新字段时会触发一次额外扫描。若源文件字段频繁变化，建议显式提供 schema 或使用 cloudFiles.schemaHints，减少推断开销和不确定性。

完成本节三条路径的摄取后，Bronze 层就具备了“来源可追溯、批次可重放、增量可续跑”的基础能力。下一步的 Silver 清洗与 Gold 聚合，都将建立在这批结构统一、元数据完整的 Bronze 表之上。