4、读写数据基础：批式写入（Append/Overwrite）、读取 Delta 表为 DataFrame、Schema 自动演化

好，咱们进入正题。Delta Lake 的读写操作，说白了就是两个动作：把数据放进去，再把数据拿出来。但这里面的门道，比普通的文件读写要多得多。

我刚开始用 Delta Lake 的时候，觉得这不就是个带事务的 Parquet 嘛。后来在项目中踩了几个坑，才明白它的设计有多讲究。今天咱们就聊聊最基础的批式读写，以及一个很实用的特性——Schema 自动演化。

4.1 批式写入：Append 与 Overwrite

批式写入，就是一次性把一批数据写进去。Delta Lake 支持两种模式：Append（追加）和 Overwrite（覆盖）。

4.1.1 Append 模式

Append 模式，顾名思义，就是把新数据追加到现有表后面。不会动已有的数据。

代码很简单，看这个例子：

# 用 DataFrame 写入 Delta 表，Append 模式
df_new_data.write \
  .format("delta") \
  .mode("append") \
  .save("/path/to/delta-table")

# 或者写入已注册的表
df_new_data.write \
  .format("delta") \
  .mode("append") \
  .saveAsTable("my_delta_table")

嗯，这里要注意：Append 模式不会检查数据是否重复。如果你不小心跑了两次同样的任务，数据就会重复。我在项目中就吃过这个亏——有一次调度系统重跑了任务，结果日志表里多了两倍的数据，排查了半天才发现。

4.1.2 Overwrite 模式

Overwrite 模式，就是覆盖整个表或分区。它会先删掉旧数据，再写入新数据。

# 覆盖整个表
df_full_data.write \
  .format("delta") \
  .mode("overwrite") \
  .save("/path/to/delta-table")

# 只覆盖特定分区
df_partition_data.write \
  .format("delta") \
  .mode("overwrite") \
  .option("replaceWhere", "date = '2024-01-15'") \
  .save("/path/to/delta-table")

Overwrite 模式最常用的场景是全量刷新。比如每天凌晨跑一次全量数据，把昨天的表整个替换掉。

我个人习惯用 replaceWhere 做分区级覆盖。为什么呢？因为全表覆盖太粗暴了，万一写失败了，整个表都不可用。分区级覆盖至少能保证其他分区不受影响。

4.2 读取 Delta 表为 DataFrame

读取 Delta 表，比写要简单得多。但有几个小技巧，能让你用起来更顺手。

4.2.1 基本读取

# 从路径读取
df = spark.read.format("delta").load("/path/to/delta-table")

# 从表名读取
df = spark.sql("SELECT * FROM my_delta_table")

# 或者用 catalog
df = spark.table("my_delta_table")

这三种方式效果一样。我个人偏好用 spark.table()，因为代码更简洁。

4.2.2 时间旅行读取

Delta Lake 最酷的特性之一，就是能读取历史版本的数据。这在数据回溯、审计、调试时特别有用。

# 按版本号读取
df_version = spark.read.format("delta") \
  .option("versionAsOf", 5) \
  .load("/path/to/delta-table")

# 按时间戳读取
df_time = spark.read.format("delta") \
  .option("timestampAsOf", "2024-01-15 10:00:00") \
  .load("/path/to/delta-table")

我记得有一次，业务方说昨天的报表数据不对。我直接用时间旅行查了前天的数据，对比了一下，发现是上游数据源出了问题。前后花了不到10分钟就定位了问题。要是没有时间旅行，得从备份里恢复数据，那可就麻烦了。

4.2.3 增量读取

如果你只想读取新增的数据，可以用 inputFiles 或 CDC 方式。但最常用的还是结合 max() 函数做增量：

# 获取当前表的最大时间戳
max_timestamp = spark.sql("SELECT max(updated_at) FROM my_delta_table").collect()[0][0]

# 读取增量数据
df_incremental = spark.read.format("delta") \
  .load("/path/to/delta-table") \
  .filter(f"updated_at > '{max_timestamp}'")

这种方法简单粗暴，但要注意：如果数据写入有延迟，可能会漏掉一些数据。更稳妥的做法是用 Structured Streaming（后面章节会讲）。

4.3 Schema 自动演化

Schema 自动演化，是 Delta Lake 的一个杀手级特性。说白了，就是当你的数据结构和表结构不一致时，Delta Lake 能自动调整表结构来适配数据。

4.3.1 为什么需要 Schema 演化？

在实际项目中，数据源的结构经常会变。比如：

• 业务系统新增了一个字段
• 上游数据源改了字段类型
• 不同批次的数据字段不一致

如果没有 Schema 演化，这些情况都会导致写入失败。我以前用 Parquet 的时候，每次结构变更都要手动改表结构，烦得很。

4.3.2 开启 Schema 演化

默认情况下，Delta Lake 的 Schema 演化是关闭的。你需要显式开启：

# 方式一：在写入时设置 option
df.write \
  .format("delta") \
  .mode("append") \
  .option("mergeSchema", "true") \
  .save("/path/to/delta-table")

# 方式二：在 SparkSession 级别设置
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

我个人建议用方式一，因为更精确。全局开启的话，可能会不小心覆盖掉一些不该改的表。

4.3.3 Schema 演化的规则

不是所有变化都能自动演化。Delta Lake 有一套明确的规则：

变化类型	是否支持自动演化	说明
新增字段	✅ 支持	新字段会追加到 Schema 末尾
删除字段	❌ 不支持	需要手动 ALTER TABLE
字段类型升级	✅ 部分支持	如 int → long, float → double
字段类型降级	❌ 不支持	如 long → int 会报错
字段重命名	❌ 不支持	会被视为新增+删除
嵌套结构变更	✅ 支持	Struct 内的字段新增

4.3.4 实际案例

假设你有一个用户表，原本只有 id、name、age 三个字段。某天业务方说要在表里加一个 email 字段。

传统做法：

1. 停掉写入任务
2. ALTER TABLE 加字段
3. 重新启动写入任务

用 Delta Lake 的 Schema 演化：

1. 直接写入带 email 字段的数据
2. Delta Lake 自动在表结构末尾追加 email 字段
3. 旧数据的 email 字段自动填充为 null

你看，省了多少事。我在项目中用过一次之后，就再也不想回到手动改 Schema 的日子了。

4.4 本章知识体系

下面这张图，帮你理清本章的核心逻辑：

4.5 小结

今天咱们聊了三个核心点：

• 批式写入：Append 和 Overwrite 两种模式，各有各的适用场景。Append 适合增量，Overwrite 适合全量。
• 读取 DataFrame：基本读取、时间旅行、增量读取，三种方式覆盖了大部分需求。
• Schema 自动演化：能自动处理新增字段和类型升级，但不能处理删除和降级。开启方式很简单，但要注意使用场景。

这些基础操作，是后续学习流式写入、Merge、事务管理的基础。你想想看，如果连读写都搞不定，后面的高级特性根本用不起来。