4. Delta Lake 基础：架构、事务日志与 ACID 保证
Delta Lake 这东西，说白了就是给数据湖加了个「数据库引擎」。我最早接触它的时候，心里也犯嘀咕：不就是个 Parquet 加了一层壳吗？后来真在项目里用上了，才发现这层壳才是精髓。
咱们先看整体架构，再拆开细聊。
4.1 Delta Lake 整体架构
Delta Lake 不是独立服务，它是个存储层中间件。跑在 Spark 或 Flink 里，读写的是 Parquet 文件，但多了一套 事务日志（Transaction Log）。
核心思路：用日志记录每次数据变更，而不是直接改文件。这样读的时候能拿到一致性快照，写的时候能冲突检测。
你看这个图，从上到下三层。计算引擎只管读写，Delta Lake 核心层负责事务管理，底层存储就是普通的文件系统。我当年第一次部署的时候，以为要装什么服务端，结果发现就是个 Jar 包，直接往 Spark 里一丢就行。
4.2 事务日志（Transaction Log）
这是 Delta Lake 最核心的部件。没有它，ACID 就是空谈。
事务日志存在 _delta_log 目录下，每个文件是一个 JSON，按版本号命名：
mnt/delta-table/
├── _delta_log/
│ ├── 00000000000000000000.json
│ ├── 00000000000000000001.json
│ ├── 00000000000000000002.json
│ └── ...
├── part-00000-xxx.snappy.parquet
├── part-00001-xxx.snappy.parquet
└── ...每个 JSON 文件记录一次原子操作。比如新增文件、删除文件、修改 schema、提交事务等。
个人习惯：我排查数据问题时，第一件事就是去翻 _delta_log 目录。看版本号有没有断层，看操作类型对不对。比查代码快多了。
4.2.1 日志条目结构
一个典型的 JSON 日志条目长这样：
{
"add": {
"path": "part-00000-xxx.snappy.parquet",
"partitionValues": {},
"size": 123456,
"modificationTime": 1700000000000,
"dataChange": true
},
"commitInfo": {
"timestamp": 1700000000000,
"operation": "WRITE",
"operationParameters": {
"mode": "Append"
}
}
}这里有几个关键字段：
- add / remove：标记文件是新增还是删除
- dataChange：是否改变了数据内容（元数据操作设为 false）
- commitInfo：记录谁、什么时间、做了什么操作
嗯，这里要注意：dataChange 这个字段很容易被忽略。我遇到过同事把 OPTIMIZE 操作误标记为 dataChange=true，结果下游流作业重复消费了全量数据。坑啊。
4.2.2 日志的原子提交
Delta Lake 怎么保证写日志不丢？靠文件系统的原子 rename 操作。
流程是这样的：
- 先写一个
.tmp临时文件 - 然后 rename 成正式的
.json文件 - 如果 rename 失败，整个事务回滚
说白了，就是利用底层文件系统的原子性。HDFS 的 rename 是原子的，S3 的 putObject 也是原子的。这招简单但有效。
我曾经踩过的坑：在 EMR 上跑 Delta Lake，S3 的 list-after-write 有最终一致性延迟。新版 Delta Lake 加了 S3ConsistencyGuard 才解决。如果你用 S3，记得升级到 1.2.0 以上。
4.3 ACID 事务保证
ACID 四个特性，Delta Lake 怎么实现的？咱们一个一个看。
| ACID 特性 | Delta Lake 实现方式 | 我见过的典型问题 |
|---|---|---|
| 原子性（Atomicity） | 事务日志原子提交 + 文件不可变 | 写一半进程挂了，重启后自动回滚 |
| 一致性（Consistency） | Schema 校验 + 约束检查 | 写入字段类型不匹配，直接拒绝 |
| 隔离性（Isolation） | 乐观并发控制（OCC） | 两个作业同时写，一个失败重试 |
| 持久性（Durability） | 数据落盘 + 日志持久化 | 机器断电，数据不丢 |
4.3.1 原子性与持久性
这两个放一起说。Delta Lake 的每次写入，要么全部成功，要么全部失败。不会出现「写了一半文件，另一半没写」的情况。
怎么做到的？
- 先把数据写到临时目录
- 然后原子提交事务日志
- 日志提交成功后，临时文件才变成正式文件
你想想看，如果写日志那一步挂了，临时文件根本不会被引用。下次读的时候，还是上一个版本的数据。
4.3.2 一致性
Delta Lake 强制 Schema 校验。你往表里写数据，字段名、类型必须匹配。
-- 假设表有 id INT, name STRING
INSERT INTO my_table VALUES ('abc', 123) -- 类型不匹配，报错我习惯在写入前用 spark.read.schema() 先检查一下目标表的 schema，避免运行时才发现不匹配。
4.3.3 隔离性（乐观并发控制）
这是 ACID 里最复杂的部分。Delta Lake 用的是 乐观并发控制（OCC），不是悲观锁。
流程：
- 读当前版本的事务日志
- 在内存中做数据变更
- 尝试提交新版本日志
- 如果版本号冲突（别人先提交了），当前事务重试
关键点：冲突检测是在提交阶段做的，不是读写阶段。所以读操作永远不会被阻塞。写操作冲突了，自动重试，对用户透明。
我遇到过最典型的场景：两个流作业同时往一张表写数据，其中一个频繁报冲突重试。后来发现是 checkpoint 间隔太短，日志文件太多，冲突概率上升。调大 checkpoint 间隔后就好了。
4.4 Checkpoint 机制
事务日志越来越多，总不能每次都从头读到尾吧？Checkpoint 就是用来解决这个问题的。
每 delta.checkpointInterval 个版本（默认 10），Delta Lake 会生成一个 Parquet 格式的 checkpoint 文件。里面记录了到当前版本为止，所有活跃文件的状态。
_delta_log/
├── 00000000000000000000.json
├── ...
├── 00000000000000000010.json
├── 00000000000000000010.checkpoint.parquet ← 快照
├── 00000000000000000011.json
└── ...读数据时，先读最近的 checkpoint，再读 checkpoint 之后的增量日志。这样就不用从头解析所有 JSON 了。
我建议：如果表的数据量很大，把 checkpoint 间隔调小一点（比如 5），能加快读性能。但写操作会多一步生成 checkpoint 的开销。需要根据实际场景权衡。
4.5 时间旅行（Time Travel）
既然所有变更都记在日志里，那回到任意历史版本就顺理成章了。
Delta Lake 支持三种回溯方式：
| 方式 | 语法 | 适用场景 |
|---|---|---|
| 按版本号 | VERSION AS OF 5 |
知道具体版本号时 |
| 按时间戳 | TIMESTAMP AS OF '2024-01-01' |
知道大概时间点时 |
| 按字符串 | @v5 或 @20240101 |
Spark SQL 快捷写法 |
-- 按版本号回溯
SELECT * FROM my_table VERSION AS OF 5
-- 按时间戳回溯
SELECT * FROM my_table TIMESTAMP AS OF '2024-01-01 12:00:00'
-- Spark SQL 快捷写法
SELECT * FROM my_table@v5时间旅行不是无限保留的。Delta Lake 默认保留 30 天的日志。超过这个时间，旧版本的数据文件和日志会被 VACUUM 清理掉。
我曾经犯过的错：有一次跑 VACUUM 忘了设保留期，把 7 天前的数据全清了。业务方要回溯数据，我只好从备份恢复。现在我的 VACUUM 脚本里永远带着 RETAIN 168 HOURS（7天），不敢偷懒。
4.6 小结
Delta Lake 的架构其实不复杂：事务日志记录一切，Checkpoint 加速读取，OCC 保证并发安全。你想想看，这些机制组合起来，就把一个普通的文件目录变成了一个有 ACID 保证的数据湖表。