📚 课程目录第4章 Delta Lake实战

4、Delta Lake实战：使用Delta Lake进行数据读写、更新、删除与合并操作

好，咱们进入实战环节。前面聊了那么多Lakehouse和Delta Lake的概念，今天直接上手操作。Delta Lake最核心的价值，就是让Spark上的数据有了“事务”和“版本”的概念。说白了，它让数据湖不再是个“只写不读”的黑洞。

我个人习惯把Delta Lake看作是在Parquet文件上加了一层“元数据日志”。每次读写操作，都会先检查这个日志，确保数据一致性。嗯，咱们就从最基础的读写开始，一步步深入到ACID操作。

核心要点：Delta Lake的所有操作都基于“事务日志”。每次写入、更新、删除，都会生成一个新的版本。这既是它的优势（时间旅行），也是需要关注的地方（小文件问题）。

4.1 环境准备与数据写入

先搭个环境。我假设你已经有Spark集群了，只需要引入Delta Lake的依赖包。如果你用Databricks，那是开箱即用。如果是自建集群，记得加上Spark Packages。

# 启动Spark Shell时引入Delta Lake
spark-shell --packages io.delta:delta-core_2.12:2.4.0

# 或者在PySpark中
pyspark --packages io.delta:delta-core_2.12:2.4.0

写数据很简单，跟写Parquet几乎一样，只是格式换成“delta”。

# Python示例
df = spark.range(0, 1000).toDF("id")
df.write.format("delta").mode("overwrite").save("/tmp/delta-table")

# 追加写入
df2 = spark.range(1000, 2000).toDF("id")
df2.write.format("delta").mode("append").save("/tmp/delta-table")

这里有个小细节：mode("overwrite")在Delta Lake里不是直接删文件，而是创建一个新版本。旧版本的数据还在，只是被标记为“历史”。我刚开始用的时候没注意，以为覆盖了就找不回来了，后来才发现可以通过时间旅行恢复。

小技巧：如果你确定旧数据不再需要，可以用VACUUM命令清理历史版本，释放存储空间。但记得保留至少一个版本用于回滚。

4.2 数据读取与时间旅行

读数据更直接。Delta Lake会自动读取最新版本的数据。

df = spark.read.format("delta").load("/tmp/delta-table")
df.show(5)

但Delta Lake真正的杀手锏是“时间旅行”。你可以读取任意历史版本的数据。这在数据回溯、审计、修复错误时特别有用。

# 读取版本0的数据
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load("/tmp/delta-table")

# 读取某个时间点的数据
df_time = spark.read.format("delta").option("timestampAsOf", "2024-01-15 10:00:00").load("/tmp/delta-table")

我曾经在项目中遇到过一个问题：某天凌晨的ETL任务跑错了，把几百万条用户数据写乱了。如果是传统数据湖，只能从备份恢复，耗时几个小时。但用Delta Lake，一行代码就回滚到了前一天的正确版本，整个过程不到5分钟。嗯，这就是时间旅行的价值。

4.3 更新与删除操作

Delta Lake支持标准的UPDATE和DELETE操作，但语法跟SQL略有不同。它使用的是Spark的DataFrame API或SQL扩展。

# 使用SQL语法
spark.sql("UPDATE delta.`/tmp/delta-table` SET id = id + 100 WHERE id < 500")

# 使用DataFrame API
from delta.tables import DeltaTable
deltaTable = DeltaTable.forPath(spark, "/tmp/delta-table")
deltaTable.update(
    condition = "id < 500",
    set = { "id": "id + 100" }
)

删除操作类似：

# SQL
spark.sql("DELETE FROM delta.`/tmp/delta-table` WHERE id > 1500")

# DataFrame API
deltaTable.delete("id > 1500")

你可能会问：这些操作真的能保证原子性吗？答案是肯定的。Delta Lake在底层会把整个操作包装成一个事务。如果中途失败，事务日志不会提交，数据状态保持不变。我测试过，在更新过程中强行kill掉Spark任务，数据不会损坏，也不会出现“半更新”的状态。

注意：UPDATE和DELETE操作在底层会重写受影响的Parquet文件。如果更新条件涉及大量数据，会产生较大的写放大。建议定期执行OPTIMIZE命令合并小文件。

4.4 合并操作（MERGE / Upsert）

合并操作是Delta Lake最常用的功能之一，尤其是在CDC（变更数据捕获）场景中。它允许你根据条件判断：如果数据存在就更新，不存在就插入。这就是所谓的“Upsert”。

# 准备源数据
source_df = spark.createDataFrame([
    (1, "Alice", 30),
    (2, "Bob", 25),
    (3, "Charlie", 35)
], ["id", "name", "age"])

# 目标表
target_table = DeltaTable.forPath(spark, "/tmp/delta-table")

# 执行合并
target_table.alias("target") \
    .merge(
        source_df.alias("source"),
        "target.id = source.id"
    ) \
    .whenMatchedUpdate(set={
        "name": "source.name",
        "age": "source.age"
    }) \
    .whenNotMatchedInsert(values={
        "id": "source.id",
        "name": "source.name",
        "age": "source.age"
    }) \
    .execute()

这段代码的逻辑很清晰：如果ID匹配，就更新name和age；如果不匹配，就插入新行。整个过程是原子性的，不会出现重复数据。

我记得有一次做实时数仓，上游的MySQL binlog每秒产生几千条变更。如果用传统方式，需要先查再判断，然后决定是更新还是插入，性能很差。换成Delta Lake的MERGE后，代码简洁了，性能也提升了3倍以上。

4.5 合并操作的进阶用法

除了基本的Upsert，MERGE还支持更复杂的场景。比如，你可以只更新某些字段，或者删除匹配的数据。

# 更复杂的合并：匹配时删除
target_table.alias("target") \
    .merge(source_df.alias("source"), "target.id = source.id") \
    .whenMatchedDelete(condition = "source.age > 40") \
    .whenNotMatchedInsertAll() \
    .execute()

这个例子中，如果源数据的age大于40，匹配到的目标行会被删除。这在数据清理场景中很实用。

另外，你还可以使用whenNotMatchedBySource子句，处理“目标表有但源表没有”的数据。这在全量同步时特别有用。

target_table.alias("target") \
    .merge(source_df.alias("source"), "target.id = source.id") \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .whenNotMatchedBySourceDelete() \
    .execute()

这段代码实现了“全量同步”：目标表与源表完全一致，多出来的数据会被删除。嗯，这就是所谓的“镜像同步”。

4.6 实战中的避坑指南

说了这么多优点，也得聊聊坑。我在项目中踩过的几个雷，分享给你：

小文件问题：频繁的UPDATE和MERGE会产生大量小文件。我曾经有个表，每天更新几百万行，一周后文件数从100个涨到了5000个，查询性能直线下降。解决方案是定期执行OPTIMIZE和ZORDER BY。
并发写入冲突：多个任务同时写同一个Delta表时，可能会遇到并发冲突。Delta Lake默认使用乐观锁，如果冲突发生，任务会重试。但重试次数过多会影响性能。我建议合理规划写入任务，避免高频并发。
事务日志膨胀：每次操作都会生成一条日志记录。如果操作非常频繁，事务日志会变得很大。建议设置delta.logRetentionDuration参数，自动清理过期日志。

我的建议：对于生产环境，建议设置以下参数：spark.databricks.delta.optimizeWrite.enabled=true（自动优化写入），spark.databricks.delta.autoCompact.enabled=true（自动合并小文件）。这两个参数能省去很多手动维护的工作。

4.7 本章知识体系

为了让你更直观地理解Delta Lake的操作体系，我画了一张流程图。它展示了从数据写入到合并操作的核心路径。

4.8 总结一下

Delta Lake的读写、更新、删除和合并操作，本质上都是在“事务日志”的协调下完成的。你不需要关心底层的文件管理，只需要关注业务逻辑。嗯，这就是Lakehouse的魅力——既有数据湖的灵活性，又有数据仓库的可靠性。

在实际项目中，我建议你从简单的读写开始，逐步尝试时间旅行和MERGE操作。遇到问题时，多看看事务日志的内容，它能告诉你Delta Lake到底在背后做了什么。