Delta Lake 事务日志：深入理解 _delta_log

聊到 Delta Lake，最核心的东西就是事务日志。我经常跟团队里的人说，搞懂了 _delta_log，你就掌握了 Delta Lake 的命脉。它不像 Hive 那种只是存个元数据，也不像普通 Parquet 文件那样只管数据。Delta Lake 的 ACID 特性、时间旅行、增量读取，全都靠这个日志来支撑。

说白了，_delta_log 就是一个目录，里面放着一堆 JSON 文件和 Parquet 文件。它们记录了这张表从创建到现在，每一次数据变更的“账本”。你想想看，如果没有这个账本，你怎么知道哪些数据是有效的？哪些是已经被删掉的？

事务日志的结构

我们先看看 _delta_log 长什么样。在 Spark 里执行 df.write.format("delta").save("/path/to/table") 之后，你会在表目录下看到这样一个结构：

/path/to/table/
├── _delta_log/
│   ├── 00000000000000000000.json
│   ├── 00000000000000000001.json
│   ├── 00000000000000000002.json
│   ├── ...
│   ├── 00000000000000000010.checkpoint.parquet
│   └── _last_checkpoint
└── data/
    ├── part-00000-xxx.snappy.parquet
    └── part-00001-xxx.snappy.parquet

每个 JSON 文件的名字是一个 20 位的数字，从 0 开始递增。这个数字就是版本号，也叫 Commit ID。我刚开始接触的时候，觉得 20 位有点夸张，后来想想，这是为了跟 Delta Lake 的协议版本兼容，也就理解了。

JSON 文件里记录了什么？每一次提交的完整元数据。包括：

• 协议版本：当前表使用的 Delta 协议版本号
• 元数据：表的 schema、分区列、配置信息
• 文件列表：这次提交新增或删除的数据文件
• 事务信息：应用 ID、操作类型、时间戳等

举个例子，一个典型的 JSON 提交文件内容大致是这样的：

{
  "protocol": {"minReaderVersion": 1, "minWriterVersion": 2},
  "metaData": {
    "id": "xxx-xxx-xxx",
    "format": {"provider": "parquet"},
    "schemaString": "{\"type\":\"struct\",\"fields\":[...]}",
    "partitionColumns": ["dt"],
    "configuration": {}
  },
  "add": {
    "path": "part-00000-xxx.snappy.parquet",
    "partitionValues": {"dt": "2024-01-01"},
    "size": 123456,
    "modificationTime": 1704067200000,
    "dataChange": true
  }
}

JSON 与 Parquet 文件的作用

你可能要问，为什么既有 JSON 又有 Parquet？这不是重复了吗？

嗯，这里有个设计上的考量。JSON 文件是“细粒度”的记录，每次提交都会生成一个。但一张表运行久了，可能有几千甚至几万个 JSON 文件。每次读元数据都要扫描所有 JSON，性能肯定扛不住。

所以 Delta Lake 引入了检查点（Checkpoint）机制。检查点文件是 Parquet 格式的，它把截止到某个版本的所有 JSON 信息“合并”成一个快照。这样，读取元数据时只需要读最新的检查点文件，再加上后面几个增量 JSON，速度就快多了。

特性	JSON 文件	Parquet 检查点文件
格式	文本，可读	列式二进制，高效
内容	单次提交的完整元数据	截止到某版本的所有文件列表快照
大小	较小（KB 级）	较大（MB 级，取决于文件数）
用途	记录每次变更，支持时间旅行	加速元数据读取，减少 JSON 扫描
生成频率	每次写入操作	每 10 个版本默认生成一次

我个人建议，在排查历史变更时，直接看 JSON 文件更直观。但如果你要快速了解当前表有哪些文件，那就用检查点文件。

提交（Commit）机制

提交机制是 Delta Lake 实现 ACID 的关键。它的核心思想是“乐观并发控制”。

流程是这样的：

1. 准备阶段：客户端读取当前最新版本的事务日志，获取表的元数据和文件列表。
2. 计算阶段：客户端根据操作类型（INSERT、UPDATE、DELETE、MERGE），计算出要新增和删除的文件列表。
3. 写入阶段：客户端将数据文件写入表目录下的 data 文件夹。
4. 提交阶段：客户端尝试在 _delta_log 目录下创建一个新的 JSON 文件，文件名是当前版本号 + 1。

这里有个关键点：文件创建是原子操作。Delta Lake 利用底层文件系统的原子性（比如 HDFS 的 rename 操作，或者 S3 的 putObject 的最终一致性），确保同一个版本号只有一个客户端能成功创建文件。如果两个客户端同时尝试创建同一个版本号的 JSON 文件，只有一个会成功，另一个会失败并重试。

提交失败后，客户端会重新读取最新版本，重新计算，再尝试提交。这个过程会一直重试，直到成功或达到最大重试次数。

检查点（Checkpoint）机制

检查点机制是为了解决“元数据膨胀”问题。你想想看，一张表运行了 1000 次写入，就有 1000 个 JSON 文件。每次读元数据都要解析这 1000 个文件，性能肯定不行。

检查点的工作原理：

• 默认每 10 个版本生成一次检查点（可通过 delta.checkpointInterval 配置）
• 检查点文件是一个 Parquet 文件，包含了截止到当前版本的所有文件列表
• 生成检查点时，Delta Lake 会读取所有 JSON 文件，合并出完整的文件列表，然后写入 Parquet
• 检查点文件的名字格式是 00000000000000000010.checkpoint.parquet

读取元数据时，Delta Lake 会先找到最新的检查点文件，然后只读取检查点之后的 JSON 文件。这样，即使有 1000 个版本，也只需要读 1 个 Parquet 文件 + 最多 9 个 JSON 文件。

另外，_last_checkpoint 文件记录了最新的检查点版本号。这个文件很小，只有几行 JSON，作用是让 Delta Lake 快速定位到最新的检查点，不用扫描整个 _delta_log 目录。

整体流程示意图

下面这张图展示了事务日志的核心工作流程，从写入操作到检查点生成：

避坑指南

最后，分享几个我在项目中遇到的真实问题：

• 检查点文件损坏：有一次集群异常重启，导致检查点文件写了一半。Delta Lake 读取时会自动回退到上一个检查点，但如果你手动删除了旧的检查点，那就麻烦了。所以，不要手动删除 _delta_log 下的任何文件。
• JSON 文件过多：一张表跑了 5000 个版本，没有生成检查点（因为配置了 delta.checkpointInterval = 1000）。结果每次查询都要等 30 秒解析 JSON。后来我手动触发了一次检查点，问题解决。
• 并发写入冲突：两个流作业同时写入同一张表，其中一个频繁失败。原因是它们使用了相同的 appId，导致事务冲突。解决方案是给每个作业分配唯一的 appId。