写入一半失败了，数据乱了怎么办？——ACID事务和Exactly-Once怎么保证？
先说结论
ACID是什么：Atomicity原子性、Consistency一致性、Isolation隔离性、Durability持久性。数据湖通过事务日志加乐观并发控制实现.
原子性实现：每次事务写一个新的日志文件如000006.json，记录添加哪些文件删除哪些文件，文件系统的原子写保证要么全成功要么全失败.
Exactly-Once保证：通过幂等写入加事务日志实现。即使任务重试也不会重复写入数据。Delta Lake的MERGE INTO、foreachBatch配合checkpoint可以实现.
最佳实践：用MERGE INTO代替INSERT OVERWRITE、启用checkpoint、监控事务日志、定期VACUUM清理旧文件.
先把这段讲顺；面试官追问时，再展开下面的原理和细节。
详细讲解
这道题面试官问得很多，因为ACID事务是数据湖和传统Hive的核心区别。我在某出海硬件品牌遇到过Spark任务写到一半失败，导致数据不一致的问题。很多人只知道数据湖支持ACID，但说不清楚原子性怎么实现、Exactly-Once怎么保证。面试官想听的是你能说清楚原理和实战经验.
真实案例：写入失败导致数据不一致
2022年在某知名出海硬件品牌，我们上线了一个用户标签更新项目，每天凌晨用Spark任务更新user_tags表.
运行一周后，业务部门反馈标签数据对不上，有些用户的标签是昨天的，有些是今天的.
排查发现：Spark任务在写数据阶段失败了，导致部分分区有新数据，部分分区有旧数据，查询时新旧行数据混杂，结果不一致.
// 问题代码：没有事务保护
val newData = spark.read.parquet("/data/user_tags/dt=2024-03-18")
.join(mysqlTags, "user_id", "full_outer")
// 直接写入，没有事务保护
newData.write.mode("overwrite").partitionBy("dt")
.save("/data/user_tags") // 如果这里失败，数据就不一致了解决方案：我们改用了Delta Lake的ACID事务：先写入临时表staging table，再用MERGE INTO原子性地替换目标表.
// 正确做法：用临时表加MERGE INTO
// 1) 写入临时表
newData.write.mode("overwrite")
.saveAsTable("user_tags_staging")
// 2) 用MERGE INTO原子性地替换目标表
spark.sql("""
MERGE INTO user_tags t
USING user_tags_staging s
ON t.user_id = s.user_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")ACID的四个特性详解
原子性Atomicity：事务是不可分割的单元，要么全部执行，要么全部不执行。Delta Lake通过事务日志实现：每次事务写一个新的JSON文件，文件系统的原子写保证写JSON文件这个操作是原子的.
一致性Consistency：事务执行前后，数据库都处于一致状态。Delta Lake通过Schema约束和事务验证保证：写入的数据必须符合表Schema.
隔离性Isolation：并发事务之间互相隔离。Delta Lake通过Serializable隔离级别实现：并发事务的执行结果和串行执行完全一致.
持久性Durability：事务一旦提交，修改就永久保存。Delta Lake通过写两次实现：先写数据文件Parquet，再写事务日志JSON.
Exactly-Once精确一次语义
Exactly-Once是流式处理的核心需求：每条数据只处理一次，不重复不丢失。Structured Streaming通过幂等写入加checkpoint可以实现Exactly-Once.
// Structured Streaming的Exactly-Once实现
val kafkaDF = spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", "localhost:9092")
.option("subscribe", "user_events")
.load()
// 用foreachBatch加幂等写入实现Exactly-Once
val query = kafkaDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
batchDF.write.format("delta")
.mode("overwrite")
.option("replaceWhere", s"batch_id = $batchId")
.save("/data/user_events")
}.option("checkpointLocation", "/checkpoint/user_events")
.start()幂等写入的核心：每次写入都带一个唯一的幂等键如batch_id、timestamp，写入前检查是否已存在，如果存在就跳过。这样即使任务重试，也不会重复写入.
实战经验：我们在出海硬件品牌用foreachBatch加replaceWhere实现Exactly-Once：将Kafka数据写入Delta Lake，用batch_id作为幂等键。改后没有再出现数据重复的问题.
ACID vs BASE对比
| 维度 | ACID传统数据库 | BASE数据湖 | Delta Lake |
|---|---|---|---|
| 原子性 | 强保证锁机制 | 最终一致 | 强保证事务日志 |
| 一致性 | 强一致立即一致 | 最终一致 | 强一致Serializable |
| 性能 | 低锁开销大 | 高无锁 | 中冲突时重试 |
| 适用场景 | 金融交易订单 | 日志数据用户行为 | 分析型查询加数据更新 |
面试高频追问
追问一：Delta Lake的原子性是怎么实现的？
Delta Lake通过事务日志加文件系统原子写实现原子性。每次事务写一个新的JSON文件，记录添加哪些文件删除哪些文件。文件系统保证写JSON文件是原子的，要么成功写入完整的文件，要么完全没写.
追问二：什么情况下会丢失数据？
三种情况会丢失数据：Executor写数据失败但Driver认为成功；使用INSERT OVERWRITE而不是MERGE INTO；checkpoint被误删。最佳实践是：用MERGE INTO、定期备份checkpoint、监控事务日志.
追问三：Exactly-Once和At-Least-Once有什么区别？
At-Least-Once保证每条数据至少处理一次，可能重复；Exactly-Once保证每条数据只处理一次，不重复不丢失。实现Exactly-Once需要：幂等写入、checkpoint、事务支持.
实战踩坑：数据重复导致报表错误
2023年在某出海硬件品牌，我们遇到过数据重复导致报表错误的问题。用Structured Streaming消费Kafka数据写入Delta Lake表.
运行一个月后，业务部门发现某些事件的计数是实际值的2倍.
排查发现：任务因为Executor OOM失败过3次，每次重启后都从上次成功的offset重新处理，导致部分数据被处理了4次.
解决方案：我们改用了幂等写入：在数据中增加batch_id列，用replaceWhere模式写入，保证同一batch_id的数据只写入一次.
// 改进后的代码：幂等写入
val query = kafkaDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
val batchWithId = batchDF.withColumn("batch_id", lit(batchId))
batchWithId.write.format("delta")
.mode("overwrite")
.option("replaceWhere", s"batch_id = $batchId")
.save("/data/user_events")
}.option("checkpointLocation", "/checkpoint/user_events")
.start()改进效果：改后没有再出现数据重复的问题，报表数据准确了.
总结
ACID事务是数据湖和传统Hive的核心区别，通过事务日志加乐观并发控制实现。原子性保证写入要么全成功要么全失败，Exactly-Once保证流式处理不重复不丢失。最佳实践是：用MERGE INTO代替INSERT OVERWRITE、启用checkpoint、使用幂等写入。面试时能说清楚ACID的四个特性、原子性怎么实现、Exactly-Once怎么保证，就很稳.
下一篇聊一个很多人担心的问题：表结构变了，历史数据怎么办？——Schema演化和向前兼容怎么实现？
你们生产环境遇到过数据不一致的问题吗？评论区聊聊～
关注「Wood聊大数据」| 9年大数据老兵，聊技术、聊职场、聊成长