📚 课程目录第3章 Delta Lake基础

3. Delta Lake基础：表创建、读写操作、事务日志、版本管理、时间旅行

3.1 核心原理：Delta Lake 的“表”到底是什么

在深入操作之前，必须先建立正确的认知模型。Delta Lake 表并不是一个全新的存储系统，而是构建在分布式文件系统（如 HDFS、S3、OSS）之上的一个元数据与数据组织层。一张 Delta 表在物理上由一个数据文件目录（存放 Parquet 数据文件）和一个名为 _delta_log 的事务日志目录共同组成。

事务日志是 Delta Lake 的灵魂。每一次对表的写操作（INSERT、UPDATE、DELETE、MERGE）都会在 _delta_log 下生成一个编号递增的 JSON 文件（如 00000000000000000000.json、00000000000000000001.json），该文件记录了本次操作的事务信息：哪些数据文件被添加、哪些被删除、表结构是否变化、操作时间等。这种设计使得 Delta Lake 具备了 ACID 事务能力——多个写入者并发操作时，由日志的原子性保证一致性。

上图展示了 Delta 表的核心结构。左侧是表目录，右侧是事务日志的展开内容。注意版本 1 的 UPDATE 操作：它并没有修改原始文件 part-0000.parquet，而是新增了一个 part-0003.parquet，并在日志中标记旧文件为“已删除”。这种不可变数据 + 日志记录变更的模式，正是时间旅行能够实现的基础。

3.2 方法流程：建表、写入与读取

在 Spark 中使用 Delta Lake 时，核心 API 与 DataFrame 操作几乎一致，只需指定格式为 delta。下面演示最基础的建表与读写流程。

3.2.1 创建 Delta 表

推荐使用 Spark SQL 直接建表，这样表结构会同步注册到 Hive Metastore，后续可通过 SQL 直接访问。

-- 使用 Spark SQL 创建 Delta 表
CREATE TABLE IF NOT EXISTS user_events (
    user_id      INT,
    event_type   STRING,
    event_time   TIMESTAMP,
    event_data   MAP<STRING, STRING>
)
USING DELTA
LOCATION '/data/delta/user_events'
PARTITIONED BY (dt STRING);

如果使用 PySpark DataFrame API，则通过 saveAsTable 或 write.format("delta").save() 创建。两者的区别在于：saveAsTable 会注册元数据，而 save() 仅写入路径。

3.2.2 写入数据

# PySpark 示例：追加写入
df = spark.createDataFrame([
    (1, "click", "2024-01-01 10:00:00", {"page": "home"}),
    (2, "view",  "2024-01-01 10:05:00", {"page": "product"})
], ["user_id", "event_type", "event_time", "event_data"])

df.write.format("delta") \
    .mode("append") \
    .partitionBy("dt") \
    .save("/data/delta/user_events")

写入模式（mode）需要特别注意：append 追加新数据；overwrite 会替换整个表或分区；errorifexists（默认）在表已存在时抛错。Delta Lake 还支持 merge 操作，用于按条件更新或插入，这是实现 CDC（Change Data Capture）的关键。

3.2.3 读取数据

# 读取最新版本
df = spark.read.format("delta").load("/data/delta/user_events")

# 读取指定版本（时间旅行）
df_v1 = spark.read.format("delta") \
    .option("versionAsOf", 1) \
    .load("/data/delta/user_events")

# 读取指定时间（时间旅行）
df_time = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-01 12:00:00") \
    .load("/data/delta/user_events")

时间旅行是 Delta Lake 最具价值的特性之一。它允许你通过 versionAsOf（按版本号）或 timestampAsOf（按时间戳）读取历史快照。这在数据回溯、审计、复现 bug 场景中极为有用。

3.3 案例场景：从零构建带版本管理的事件表

假设你负责一个用户行为分析平台，需要维护一张事件表，并支持以下需求：每日追加新事件；偶尔修正历史错误数据；分析师需要查看三天前的数据状态。下面演示完整流程。

场景操作序列：

第 1 天：创建表并写入 100 条事件（版本 0）。
第 2 天：追加 50 条新事件（版本 1）。
第 3 天：发现版本 0 中有 5 条事件类型错误，执行 UPDATE 修正（版本 2）。
第 4 天：分析师需要对比修正前后的数据，使用时间旅行读取版本 1 和版本 2。

-- 第 3 天：修正错误数据
MERGE INTO user_events AS target
USING (
    SELECT user_id, 'corrected_type' AS event_type, event_time, event_data, dt
    FROM user_events
    WHERE user_id IN (101, 102, 103, 104, 105)
) AS source
ON target.user_id = source.user_id AND target.event_time = source.event_time
WHEN MATCHED THEN UPDATE SET target.event_type = source.event_type;

-- 第 4 天：对比版本 1 与版本 2 的数据差异
SELECT COUNT(*) AS cnt_v1 FROM delta.`/data/delta/user_events` VERSION AS OF 1;
SELECT COUNT(*) AS cnt_v2 FROM delta.`/data/delta/user_events` VERSION AS OF 2;

执行上述操作后，通过 DESCRIBE HISTORY user_events 可以查看表的完整操作历史，包括每次操作的版本号、操作类型、操作时间、影响的行数等元数据。

版本	操作	时间	数据文件变化
0	CREATE TABLE + Append	2024-01-01 09:00	新增 2 个 Parquet 文件
1	Append	2024-01-02 09:30	新增 1 个 Parquet 文件
2	Merge (Update)	2024-01-03 14:00	删除 1 个文件，新增 1 个文件

3.4 误区与复盘：版本管理与时间旅行的常见陷阱

误区一：认为时间旅行会永久保留所有历史版本。

Delta Lake 默认保留最近 30 天的提交日志，但数据文件并不会自动清理。如果执行了 VACUUM 命令，超过保留期的历史数据文件会被物理删除，此时时间旅行将无法访问那些版本。因此，在需要长期保留历史快照的场景中，应适当调高 delta.logRetentionDuration 和 delta.deletedFileRetentionDuration 配置。

误区二：混淆“表版本”与“数据分区”。

表版本是全局递增的事务编号，而分区是物理存储的目录组织方式。一次写入操作可能涉及多个分区，但只会产生一个新版本。时间旅行是基于版本而非分区的，你不能通过指定分区来回到过去的状态。

复盘建议：在生产环境中，建议为关键表开启 Change Data Feed（TBLPROPERTIES ('delta.enableChangeDataFeed' = true)），这样每次操作不仅记录在事务日志中，还会生成标准化的变更数据行（包含 _change_type 列），方便下游流式消费。同时，定期执行 OPTIMIZE 合并小文件，并使用 ZORDER BY 优化查询性能——但要注意，OPTIMIZE 也会产生新版本，且会重写数据文件，因此执行前应确认保留策略。

最后需要强调的是，Delta Lake 的版本管理是乐观并发控制的实现：多个写入者同时提交时，后提交者若发现与先提交者存在冲突（如修改了同一文件），会失败并重试。理解这一机制有助于你设计合理的写入任务调度，避免不必要的冲突重试。