3. Delta Lake 基础：安装、时间旅行与 Schema 演进

好，咱们进入正题。Delta Lake 是什么？说白了，它就是在你的数据湖上，加了一层事务性保障。你想想看，以前用 Hive 或者 Spark 直接写 Parquet 文件，最怕什么？写到一半任务挂了，留下一堆脏数据。Delta Lake 就是来解决这个问题的。

我个人习惯把 Delta Lake 看作「数据湖的 ACID 引擎」。它用了一套事务日志（Transaction Log）来记录每一次数据变更，这才有了后面我们要讲的时间旅行、版本控制这些高级特性。

3.1 安装 Delta Lake

安装其实不复杂，但有几个坑我得先跟你说。我刚开始用的时候，就因为版本没对齐，折腾了大半天。

3.1.1 Spark 环境下的安装

如果你用的是 Spark，Delta Lake 是以一个 Maven 依赖的形式提供的。你只需要在启动 Spark 时，加上这个 package 参数：

# 启动 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

3.1.2 在代码中引入依赖

如果你是用 Maven 或 Gradle 构建项目，直接在 pom.xml 里加：

<dependency>
    <groupId>io.delta</groupId>
    <artifactId>delta-core_2.12</artifactId>
    <version>2.4.0</version>
</dependency>

嗯，这里要注意，_2.12 指的是 Scala 版本，别搞错了。

3.1.3 验证安装

装好之后，怎么确认它跑起来了？很简单，执行一行代码：

# PySpark 中验证
spark.range(0, 5).write.format("delta").save("/tmp/delta-test")
spark.read.format("delta").load("/tmp/delta-test").show()

如果能看到 0 到 4 的数字，说明 Delta Lake 已经正常工作了。

3.2 核心特性：时间旅行与版本控制

这两个特性是 Delta Lake 的杀手锏。我经常跟团队说，有了时间旅行，你就等于给数据湖装了一台「时光机」。

3.2.1 版本控制机制

Delta Lake 的每一次写操作，都会生成一个新版本。这些版本信息存储在 _delta_log 目录下，里面是一堆 JSON 文件。

举个例子，你执行了三次写操作：

# 第一次写入
df.write.format("delta").mode("overwrite").save("/data/events")

# 第二次写入
df2.write.format("delta").mode("append").save("/data/events")

# 第三次写入
df3.write.format("delta").mode("overwrite").save("/data/events")

这时候，/data/events/_delta_log/ 目录下会有三个文件：00000000000000000000.json、00000000000000000001.json、00000000000000000002.json。每个文件记录了一次操作的元数据。

3.2.2 时间旅行：读取历史版本

时间旅行有两种方式：按版本号查，或者按时间戳查。

# 按版本号读取（版本 1）
spark.read.format("delta")
    .option("versionAsOf", "1")
    .load("/data/events")
    .show()

# 按时间戳读取
spark.read.format("delta")
    .option("timestampAsOf", "2024-01-15 10:00:00")
    .load("/data/events")
    .show()

为什么会需要这个功能？我遇到过好几次，业务方说「昨天的报表数据不对，好像是凌晨那批任务写错了」。有了时间旅行，我直接查凌晨之前的数据版本，一对比就知道问题出在哪。

3.2.3 数据回滚

回滚其实不是 Delta Lake 的原生操作，但我们可以通过「覆盖写入」来实现：

# 把版本 1 的数据覆盖到当前版本
val df = spark.read.format("delta")
    .option("versionAsOf", "1")
    .load("/data/events")

df.write.format("delta").mode("overwrite")
    .option("replaceWhere", "true")
    .save("/data/events")

3.3 Schema 演进：让表结构活起来

传统数据仓库里，改表结构是个大工程。但在 Delta Lake 里，Schema 演进是默认支持的。说白了，你加一列、改一列类型，Delta Lake 会自动帮你处理。

3.3.1 自动 Schema 演进

当你写入的数据和现有表的 Schema 不一致时，Delta Lake 可以自动合并：

# 原始表只有 name 和 age 两列
df_old = spark.createDataFrame([("Alice", 30)], ["name", "age"])
df_old.write.format("delta").mode("overwrite").save("/data/users")

# 新数据多了一列 city
df_new = spark.createDataFrame([("Bob", 25, "Beijing")], ["name", "age", "city"])

# 开启 Schema 演进后写入
df_new.write.format("delta").mode("append")
    .option("mergeSchema", "true")
    .save("/data/users")

# 现在表里有三列了
spark.read.format("delta").load("/data/users").printSchema()

3.3.2 手动管理 Schema

有时候自动演进太「智能」了，反而会带来问题。比如你写错了列名，Delta Lake 会直接加一列，而不是报错。所以我建议，在关键生产任务中，手动控制 Schema 变更：

# 手动添加列
spark.sql("""
    ALTER TABLE events ADD COLUMNS (source string AFTER timestamp)
""")

# 修改列注释
spark.sql("""
    ALTER TABLE events ALTER COLUMN user_id COMMENT '用户唯一标识'
""")

3.4 核心概念关系图

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

3.5 实践中的注意事项

最后，分享几个我在项目中积累的经验：

• 事务日志会膨胀：每次操作都生成一个 JSON 文件，时间久了，_delta_log 目录下可能有成千上万个文件。我建议定期执行 VACUUM 清理过期版本。
• 时间旅行不是无限期的：默认保留 30 天的历史版本。如果你需要更长的回溯周期，记得调整 delta.logRetentionDuration 参数。
• Schema 演进要谨慎：自动演进虽然方便，但容易引入脏数据。我的做法是：开发环境全开，测试环境半开，生产环境全关。