4、批处理读写实践：使用 DataFrame 进行批量写入、覆盖与追加模式、分区表操作
好，咱们进入实操环节。批处理读写是 Delta Lake 最基础、也是最常用的能力。说白了，你每天跑的那些离线任务，十有八九都是批处理。我个人习惯把批处理比作「搬砖」—— 一次性把一堆数据从一个地方搬到另一个地方，搬完就完事。但 Delta Lake 这块砖，搬起来有点讲究。
4.1 批量写入：从 DataFrame 到 Delta 表
先看最基础的操作。把 Spark DataFrame 写入 Delta 表，语法跟写 Parquet 几乎一样，但背后多了事务日志和版本管理。
# 从数据源读取，生成 DataFrame
df = spark.read.format("csv").option("header", "true").load("/data/raw/events.csv")
# 写入 Delta 表（默认模式：Append）
df.write.format("delta").save("/data/delta/events")
嗯，这里要注意：第一次写入时，Delta Lake 会自动创建表。你不需要提前建表，也不需要定义 schema——它会从 DataFrame 的 schema 推断出来。当然，我建议你显式指定 schema，避免类型推断翻车。我在项目中遇到过，CSV 里某个字段全是数字，Spark 自动推断成 int，结果后来混进来一条带字母的数据，直接报错。
.option("mergeSchema", "false")。默认是 false，一旦 schema 不匹配就会报错，而不是偷偷改你的表结构。
4.2 追加模式 vs 覆盖模式
写入模式就两种：Append 和 Overwrite。但 Delta Lake 的 Overwrite 跟普通文件系统的覆盖不太一样，它不会真的把文件删掉再写，而是通过事务日志标记旧文件为「已删除」，新文件直接写入。这保证了原子性——要么全成功，要么全失败。
| 模式 | 语法 | 行为 | 适用场景 |
|---|---|---|---|
| Append | .mode("append") |
新数据追加到现有数据之后 | 日志追加、增量数据入湖 |
| Overwrite | .mode("overwrite") |
替换整个表或分区 | 全量刷新、数据重跑 |
# 追加模式（默认）
df.write.format("delta").mode("append").save("/data/delta/events")
# 覆盖模式
df.write.format("delta").mode("overwrite").save("/data/delta/events")
time travel 恢复，但生产环境请务必确认操作无误。我曾经手滑把全量覆盖写成了分区覆盖，还好有版本回溯救场。
4.3 分区表操作：让查询更快
分区是提高查询性能的利器。Delta Lake 的分区跟 Hive 分区类似，但更智能——它自动管理分区目录，不需要你手动添加分区。
4.3.1 创建分区表
写入时指定分区列即可。我一般按日期分区，因为大部分查询都带时间范围。
# 按日期分区写入
df.write.format("delta") \
.partitionBy("event_date") \
.mode("append") \
.save("/data/delta/events_partitioned")
你想想看，如果表有 1 亿条数据，按天分区后查询某一天的数据，Delta Lake 只需要扫描一个分区目录，而不是全表扫描。性能差距是数量级的。
4.3.2 分区覆盖：只替换特定分区
这是 Overwrite 模式的高级用法。有时候你只想重跑某一天的数据，而不是整个表。Delta Lake 支持动态分区覆盖，只替换匹配的分区。
# 只覆盖 event_date='2024-01-15' 这个分区
df_filtered = spark.read.format("delta").load("/data/delta/events_partitioned") \
.filter("event_date = '2024-01-15'")
# 重新计算后写回，只覆盖该分区
df_corrected.write.format("delta") \
.mode("overwrite") \
.option("replaceWhere", "event_date = '2024-01-15'") \
.save("/data/delta/events_partitioned")
replaceWhere 选项告诉 Delta Lake：「我只覆盖这些分区，其他分区别动」。这比全表覆盖安全得多，也快得多。
4.3.3 分区修剪（Partition Pruning）
Delta Lake 会自动利用分区信息进行查询优化。你写查询时，尽量把分区条件放在 WHERE 子句里。
# 好的写法：触发分区修剪
df = spark.read.format("delta").load("/data/delta/events_partitioned") \
.filter("event_date >= '2024-01-01' AND event_date <= '2024-01-31'")
# 不好的写法：全表扫描
df = spark.read.format("delta").load("/data/delta/events_partitioned") \
.filter("year(event_date) = 2024")
为什么第一个好？因为 Delta Lake 的查询优化器能直接定位到 1 月份的分区目录，跳过其他 11 个月的数据。第二个写法用了函数 year()，优化器没法直接推断分区范围，只能全表扫描。嗯，这里是个常见的坑。
4.4 核心流程：批处理读写决策树
下面这张图总结了批处理读写的核心决策逻辑。我画这张图时，把平时踩过的坑都融进去了。
4.5 实战避坑指南
最后分享几个我踩过的坑，希望能帮你省点 debug 时间。
- 分区列类型要一致：分区列的类型必须跟表定义一致。我曾经把日期写成 string 类型，结果分区目录名是 "2024-01-15"，后来改写成 date 类型，分区目录变成了 "event_date=2024-01-15"，查询时类型不匹配，报错。嗯，血的教训。
- 小文件问题：频繁的 Append 操作会产生大量小文件。我建议定期执行
OPTIMIZE合并小文件，或者写入时用.option("maxRecordsPerFile", 1000000)控制每个文件的大小。 - 覆盖模式别乱用：生产环境里，我习惯在 Overwrite 之前先查一下目标表的数据量，确认不是空表。如果空表被覆盖了，数据就丢了——虽然能恢复，但恢复操作本身也有成本。
- 分区数不要太多：按天分区没问题，但按小时分区就要小心了。一年 8760 个分区，元数据管理开销很大。我一般建议分区粒度到天，除非数据量特别大（每天几十 TB）。