Parquet表写入到一半失败就脏了？——数据湖的ACID事务是怎么做到的？

数据湖手册第 1 题约 13 分钟读完

本文目录

先说结论
我在出海硬件公司的真实经历
传统Parquet/Hive的问题
Delta Lake的ACID实现机制
Iceberg的ACID实现机制
Delta Lake vs Iceberg：事务实现对比
面试高频追问
实战踩坑：Delta Lake的'幽灵文件'问题
总结

先说结论

核心区别：数据湖（Delta Lake/Iceberg/Hudi）在Parquet之上加了事务日志层，把元数据和数据分开存储。

Delta Lake方案：用JSON事务日志（_delta_log）记录每次提交的操作，原子性通过'写日志+写数据+原子提交'实现。

Iceberg方案：用元数据文件（metadata.json）和清单文件（manifest）追踪数据文件，通过版本化元数据实现原子性。

关键能力：原子写入、失败回滚、并发写入时保证一致性——这就是ACID。

先把这段讲顺；面试官追问时，再展开下面的原理和细节。

详细讲解

这道题面试官问得特别多，因为ACID事务是数据湖和传统Hive表最本质的区别。我在某出海硬件品牌从Hive迁移到Delta Lake时，第一个感受到的优势就是'写入失败了不会脏数据'。很多人只回答'有事务'，但说不清楚'怎么做到的'。面试官想听的是你能说清楚事务的实现机制。

我在出海硬件公司的真实经历

2022年我在某知名出海硬件品牌负责数据平台，当时还用Hive on Parquet。有个用户复购率分析的作业，每天凌晨跑，处理约5亿条用户行为数据。有一次作业跑到一半集群宕机了，结果Parquet文件写了一半——部分分区有新数据，部分分区还是旧数据。更糟糕的是，下游报表直接基于这个'半成品'数据跑完了，给出的复购率数据完全错误，被业务部门投诉。

当时我们只能手动识别哪些分区是新的、哪些是旧的，然后删除所有新分区重新跑整个作业。这还只是'有监控发现'的情况，如果没监控，脏数据会一直影响下游。后来我们迁移到Delta Lake，这个问题彻底解决了——写入失败时，Delta Lake会自动回滚，就像什么都没发生过。

// Delta Lake写入：失败时自动回滚，不会留下脏数据
df.write
  .format("delta")
  .mode("overwrite")
  .save("/data/delta/user_behavior")
 
// 即使中途失败，读者仍然读到的是之前完整的版本
// 不会出现'部分分区新数据、部分分区旧数据'的情况

传统Parquet/Hive的问题

传统的Parquet或Hive表之所以会有'脏数据'问题，核心原因是：

① 没有事务概念：写入是直接操作文件系统，没有事务日志来记录'这次写入包含哪些文件'。

② 原子性缺失：覆盖写入时，是'先删除旧文件、再写新文件'，中间如果有失败，就卡在'旧文件已删、新文件未完成'的状态。

③ 写入顺序问题：写入多个分区时，如果分区3失败，分区1-2的新文件已经落地，读者会读到不一致的数据。

-- 传统Hive写入：分区级别原子，但表级别非原子
INSERT OVERWRITE TABLE user_behavior PARTITION(dt='2024-03-19')
SELECT * FROM source WHERE dt = '2024-03-19';
 
-- 如果这个作业在写入分区3时失败，
-- 分区1-2已经有新数据，分区3-5还是旧数据
-- 下游查询时就会读到'部分新、部分旧'的不一致状态

这个问题的本质是：文件系统本身没有事务语义，数据湖需要在文件系统之上'实现'事务能力。

Delta Lake的ACID实现机制

Delta Lake的核心创新是：在Parquet数据文件之外，单独维护一个事务日志目录（_delta_log）。每次写入都是一个'原子提交'过程。

# Delta Lake的目录结构
/data/delta/user_behavior/
├── _delta_log/           # 事务日志目录（关键！）
│   ├── 000000000.json   # 第0次提交的日志
│   ├── 000000001.json   # 第1次提交的日志
│   └── 000000002.json   # 第2次提交的日志
└── part-00001-*.parquet # 实际的Parquet数据文件

写入流程（原子性的关键）：

1. 准备阶段：将数据写入新的Parquet文件（临时文件，不影响现有读取）

2. 日志记录：生成一个新的JSON日志文件（如000000003.json），记录'这次提交添加/删除了哪些数据文件'

3. 原子提交：将新的日志文件原子性地写入_delta_log目录（文件系统的rename操作是原子的）

4. 失败处理：如果任何步骤失败，新的日志文件不会被提交，读者仍然读到旧的日志版本，新数据文件会被后续的Vacuum清理

{
  "commitInfo": {
    "timestamp": 1710835200000,
    "operation": "WRITE",
    "operationParameters": { "mode": "overwrite" }
  },
  "add": [
    { "path": "part-00001-*.parquet", "size": 123456, "stats": "..." },
    { "path": "part-00002-*.parquet", "size": 234567, "stats": "..." }
  ],
  "remove": [
    { "path": "part-00099-*.parquet", "deletionTimestamp": 1710835200000 }
  ]
}

为什么这个设计是原子的？因为日志文件是顺序编号的，读取数据时'只要读到最新的日志文件（如000000003.json），就能确定哪些数据文件是有效的'。即使写入过程中失败，旧版本日志（000000002.json）仍然有效，读者不会看到'半成品'状态。

Iceberg的ACID实现机制

Iceberg采用类似的思路，但实现方式不同。它用'版本化的元数据文件'来追踪数据，而不是单个递增的JSON日志。

# Iceberg的目录结构
/data/iceberg/user_behavior/
├── metadata/
│   ├── v1.metadata.json      # 版本1的元数据
│   ├── v2.metadata.json      # 版本2的元数据
│   ├── v3.metadata.json      # 版本3的元数据
│   ├── s1-manifest-list     # 清单列表（追踪manifest文件）
│   └── m1.avro               # 清单文件（追踪数据文件）
└── data/
    └── part-00001-*.parquet # 实际的Parquet数据文件

Iceberg的三层结构：

① 元数据文件（metadata.json）：记录表的Schema、分区规范、当前快照ID等表级信息

② 清单文件（manifest，Avro格式）：记录数据文件的路径、分区值、记录数、文件大小等

③ 清单列表（manifest list）：记录'本次快照包含哪些manifest文件'，是清单的清单

# Iceberg元数据文件（metadata.json）简化版
{
  "format-version": 2,
  "schema": { ... },
  "partition-spec": [ ... ],
  "current-snapshot-id": 1234567890,
  "snapshots": [
    {
      "snapshot-id": 1234567890,
      "timestamp-ms": 1710835200000,
      "summary": { "operation": "overwrite" },
      "manifest-list": "s1-manifest-list"
    }
  ]
}

原子性怎么保证？每次提交时，生成一个新的snapshot（包含新的manifest-list），然后更新metadata.json中的'current-snapshot-id'字段。读者读取时，通过'最新的metadata.json → current-snapshot-id → manifest-list → manifest → 数据文件'链路访问数据。只要metadata.json的更新是原子的（通过HDFS的rename或S3的原子put），整个快照的切换就是原子的。

Delta Lake vs Iceberg：事务实现对比

维度	Delta Lake	Apache Iceberg
事务日志	单个JSON文件递增（00000000.json）	多层结构（metadata → manifest list → manifest）
元数据存储	Delta Log（JSON）	Metadata File（JSON）+ Manifest（Avro）
版本追踪	日志文件序号	Snapshot ID（UUID）
原子性保证	原子追加日志文件	原子更新metadata.json的current-snapshot-id
读取路径	最新日志 → add/remove → 数据文件	metadata → snapshot → manifest → 数据文件
性能特点	日志文件会变大（需定期压缩）	Avro格式的manifest更高效，适合大规模
引擎兼容	Spark原生（Databricks最优）	Spark/Flink/Trino/Presto全支持

面试高频追问

追问一：如果两个作业同时写入同一张Delta Lake表，会怎么处理？

这是并发写入的经典场景。Delta Lake采用'乐观并发控制（Optimistic Concurrency Control）'。每个写入者在提交前会检查：'我读取时的日志版本，和现在最新的日志版本是否一致'。如果一致，说明没有其他写入者，可以提交；如果不一致，说明有其他写入者在我之前提交了，需要重试或失败。

// Delta Lake的并发冲突处理
df.write.format("delta").mode("overwrite")...
// 如果提交时发现日志版本已变，会抛出 ConcurrentModificationException
// 错误信息会告诉你：'你基于版本3写，但当前已经是版本4了'
// 解决方式：重试整个作业，或者改用merge语义

Iceberg的处理方式类似，但提供了更细粒度的冲突检测——可以配置'当冲突发生时，只对特定列报错'，允许某些场景下的并发写入。

追问二：Delta Lake的日志文件会不会越来越大？怎么处理？

是的，这是个实际问题。每次写入都会生成一个新的JSON日志文件，时间久了会有成千上万个小文件。Delta Lake提供了'日志压缩'功能：

-- Delta Lake的日志压缩（Checkpoint）
-- 会将最近的N个JSON日志合并成一个Parquet文件
CALL checkpoint_tablename('/data/delta/user_behavior');
 
-- 或者在每次写入时自动触发（推荐配置）
SET spark.databricks.delta.checkpointInterval.write = 10;
-- 每写入10次，自动生成一个checkpoint Parquet文件

Checkpoint是一个Parquet文件，包含了'从0到当前版本的所有有效数据文件的快照'。读取时，如果检测到有checkpoint，会直接从最近的checkpoint开始，只读取之后的增量JSON日志。这样既保留了历史版本，又避免了读取成千上万个JSON文件。

追问三：Time Travel功能是怎么实现的？

Time Travel依赖于'不可变的日志历史'。因为每次提交都生成新的日志文件（Delta Lake）或新的metadata.json（Iceberg），旧版本的元数据不会被删除，所以可以'回到任意历史版本'。

-- Delta Lake Time Travel：读取昨天上午10点的数据
SELECT * FROM user_behavior VERSION AS OF 3;
SELECT * FROM user_behavior TIMESTAMP AS OF '2024-03-19 10:00:00';
 
-- Iceberg Time Travel
SELECT * FROM user_behavior VERSION AS OF 1234567890;
SELECT * FROM user_behavior FOR SYSTEM_TIME AS OF
  TIMESTAMP '2024-03-19 10:00:00';

这在Hive时代是不可想象的——一旦覆盖写入，旧数据就没了。数据湖的Time Travel对于'数据回滚、历史对比、调试'非常有用。

实战踩坑：Delta Lake的'幽灵文件'问题

2023年初，我在某短剧出海公司用Delta Lake构建广告归因表。有个作业每天用overwrite模式写入全量数据（约2TB）。运行了一个月后，存储费用激增——明明表大小只有2TB，但S3存储用了6TB！

排查后发现：Delta Lake的overwrite虽然逻辑上'替换了旧数据'，但物理上旧文件还在，只是在事务日志中标记为'removed'。默认情况下，这些'被删除的文件'不会立即物理删除，需要手动运行Vacuum命令。

-- 清理7天前被删除的文件（默认保留7天）
VACUUM user_behavior RETAIN 7 HOURS;
 
-- ⚠️ 默认情况下，Delta Lake禁止Vacuum < 7天的数据
-- 这是为了防止Time Travel失败
-- 如果确实需要，可以强制关闭保留期检查（不推荐）
SET spark.databricks.delta.vacuum.retentionCheck.enabled = false;

解决方案：我们配置了自动Vacuum策略，每次写入后自动清理7天前的旧文件。同时调整了作业逻辑——改用MERGE INTO而不是overwrite，只更新变更的分区，这样每次写入只产生少量新文件，存储成本降到了2.5TB。

这个坑的本质是：'逻辑删除不等于物理删除'。数据湖为了支持Time Travel，默认会保留历史版本。如果不需要长期回滚，记得定期Vacuum。

总结

数据湖的ACID事务是'Parquet + 事务日志层'的组合创新。Delta Lake用递增JSON日志，Iceberg用版本化元数据+多层Manifest。核心都是'原子提交日志/元数据'来实现幂等和回滚。面试时能说清楚'为什么传统Hive会脏、数据湖怎么实现原子性、Time Travel原理'，这道题就稳了。

下一篇聊一个很多人用错的功能：更新一条数据要全量重写整个分区？——MERGE INTO怎么高效更新删除？

你们生产环境用Delta Lake还是Iceberg？有没有遇到过'写入失败脏数据'的情况？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长

刷题进度保存在当前浏览器