同时写入同一张表会冲突吗？——并发控制和隔离级别是什么？

-- 问题代码：两个任务都用INSERT OVERWRITE
-- 任务A：用户行为同步
INSERT OVERWRITE TABLE user_behavior
PARTITION (dt='2024-03-19', hour='10')
SELECT * FROM source_user_behavior
WHERE dt='2024-03-19' AND hour='10';
 
-- 任务B：数据质量修复
INSERT OVERWRITE TABLE user_behavior
PARTITION (dt='2024-03-19', hour='10')
SELECT * FROM user_behavior_staging
WHERE dt='2024-03-19' AND hour='10';  -- 修复后的数据
 
-- 结果：后完成的任务会覆盖先完成的任务的数据

// 乐观并发控制的基本逻辑（伪代码）
val transaction = new Transaction()
val startVersion = transaction.getCurrentVersion()  // 读取当前版本
 
// 修改数据
val newData = transformData(readData(startVersion))
 
// 尝试提交
val committed = transaction.tryCommit(newData, startVersion)
if (committed) {
  println("提交成功")
} else {
  println("冲突，重试...")
  retry()  // 自动重试
}

# Delta Lake的事务日志结构
/user/hive/warehouse/user_behavior/_delta_log/
├── 00000000000000000001.json  -- 第1次事务：创建表
├── 00000000000000000002.json  -- 第2次事务：写入数据
├── 00000000000000000003.json  -- 第3次事务：MERGE INTO
└── 00000000000000000004.json  -- 第4次事务：OPTIMIZE
 
# 每个JSON文件记录这次事务的详细信息
# {'add': ['/path/to/file1.parquet', ...],
#  'remove': ['/path/to/file2.parquet', ...]}

冲突类型：Delta Lake会检测两种冲突：1）写写冲突（Write-Write Conflict）：两个事务同时修改同一分区，后提交的会失败；2）读写冲突（Read-Write Conflict）：一个事务读取数据时，另一个事务修改了同一分区，读事务会失败。

数据湖格式并发控制对比

维度	Delta Lake	Apache Iceberg	Apache Hudi
并发控制	乐观并发控制	乐观并发控制	乐观并发控制
元数据存储	_delta_log JSON文件	Hive Metastore + 元数据文件	.hoodie元数据目录
冲突处理	自动重试（可配置次数）	手动重试	自动重试
隔离级别	Serializable（可串行化）	Serializable（可串行化）	Write Serializable（写可串行化）
INSERT OVERWRITE	支持（原子性）	支持（原子性）	支持（原子性）
MERGE INTO	支持（原子性）	支持（原子性）	支持（原子性）
我的选择	Databricks生态首选	多引擎生态首选	流式写入首选

如何避免并发冲突？

方法1：避免多个任务写入同一分区——这是最简单有效的方法。例如：任务A写入dt=2024-03-19/hour=10，任务B写入dt=2024-03-19/hour=11（错开时间）；或者任务A写入user_behavior表，任务B写入user_behavior_staging表，然后用MERGE INTO合并。

-- 错误做法：两个任务写入同一分区
-- 任务A：INSERT OVERWRITE user_behavior PARTITION (dt='2024-03-19', hour='10')
-- 任务B：INSERT OVERWRITE user_behavior PARTITION (dt='2024-03-19', hour='10')
 
-- 正确做法1：错开时间
-- 任务A：INSERT OVERWRITE ... PARTITION (dt='2024-03-19', hour='10')
-- 任务B：INSERT OVERWRITE ... PARTITION (dt='2024-03-19', hour='11')
 
-- 正确做法2：用MERGE INTO代替INSERT OVERWRITE
-- 任务B：MERGE INTO user_behavior t
-- USING staging_data s ON t.user_id = s.user_id
-- WHEN MATCHED THEN UPDATE SET ...
-- WHEN NOT MATCHED THEN INSERT *;

方法2：使用动态分区覆盖——Delta Lake的INSERT OVERWRITE可以'dynamically overwrite'，即只覆盖'有数据的分区'，不影响其他分区。这样可以减少冲突范围。

-- 动态分区覆盖（Delta Lake）
INSERT OVERWRITE TABLE user_behavior
PARTITION (dt, hour)  -- 分区列
SELECT user_id, event_type, event_time, dt, hour  -- 包含分区列
FROM source_data
WHERE dt='2024-03-19' AND hour IN ('10', '11');
-- 只会覆盖dt=2024-03-19/hour=10和hour=11的分区
-- 其他分区（如hour=12）不会被影响

方法3：监控和调优并发参数——监控'冲突重试次数'，如果重试次数太多（>5次），说明设计有问题。可以调整spark.sql.delta.commitRetry.maxAttempts（默认10次）和spark.sql.delta.commitRetry.interval（默认100ms）。

面试高频追问

追问一：Delta Lake的ACID是怎么实现的？

Delta Lake通过'事务日志+ 乐观并发控制'实现ACID：1）原子性（Atomicity）：每次事务写一个新的JSON文件（如000006.json），包含'添加哪些文件、删除哪些文件'的完整操作，文件系统的'原子写'保证'要么全成功要么全失败'；2）一致性（Consistency）：事务日志记录了所有操作，保证数据始终处于'一致状态'；3）隔离性（Isolation）：通过乐观并发控制，并发事务互相隔离；4）持久性（Durability）：事务日志和数据文件都持久化到磁盘，不会丢失。

追问二：什么是'Serializable'隔离级别？

'Serializable'是最高级别的隔离级别，保证'并发事务的结果'和'串行执行的结果'完全一致。例如：事务A读取dt=2024-03-19的数据，事务B修改同一分区的数据，如果隔离级别是'Read Committed'，事务A可能读到'脏数据'（事务B修改前的数据）；如果是'Serializable'，事务A会被阻塞或重试，保证不会读到'不一致'的数据。Delta Lake和Iceberg都是'Serializable'，Hudi是'Write Serializable'（只保证写的可串行化）。

追问三：并发冲突会自动重试吗？需要手动处理吗？

Delta Lake会自动重试（默认最多10次），如果重试10次还是失败，才会抛出异常。大部分情况下不需要手动处理，但如果'冲突频繁'，说明设计有问题（多个任务写入同一分区），需要调整ETL设计。可以通过spark.sql.delta.commitRetry.maxAttempts调整重试次数，通过spark.sql.delta.commitRetry.interval调整重试间隔。

实战踩坑：并发冲突导致ETL失败

2023年在某出海硬件品牌，我们遇到过一次'并发冲突导致ETL失败'的问题。当时我们上线了一个'实时数仓'项目，用Structured Streaming每5分钟写入一次Delta Lake表。运行一段时间后发现：

1）Streaming任务频繁失败，报错'ConcurrentAppendException'

2）查看日志：'事务冲突，重试10次后失败'

3）排查发现：多个Streaming任务同时写入同一张表（user_behavior）的同一分区

// 问题代码：多个Streaming任务同时写入同一表
// Streaming任务A：写入page_view事件
val streamA = spark.readStream.format("delta")...
streamA.writeStream.format("delta")
  .option("checkpointLocation", "/checkpoint/a")
  .partitionBy("dt", "hour")
  .start("user_behavior")  // 同一张表
 
// Streaming任务B：写入click事件
val streamB = spark.readStream.format("delta")...
streamB.writeStream.format("delta")
  .option("checkpointLocation", "/checkpoint/b")
  .partitionBy("dt", "hour")
  .start("user_behavior")  // 同一张表！
 
// 结果：两个任务写入同一分区，频繁冲突

解决方案：我们改用'Union后统一写入'——先启动多个Streaming任务读取不同数据源，然后用'union'合并，最后用一个任务写入Delta Lake。这样只有一个'写入任务'，不会冲突。

// 正确做法：Union后统一写入
val streamA = spark.readStream.format("delta").load("source_a")
val streamB = spark.readStream.format("delta").load("source_b")
 
// Union后统一写入
val stream = streamA.union(streamB)
stream.writeStream.format("delta")
  .option("checkpointLocation", "/checkpoint/combined")
  .partitionBy("dt", "hour")
  .start("user_behavior")  // 只有一个写入任务

改进效果：改后Streaming任务再也没有出现过'并发冲突'，稳定性大幅提升。

总结

并发写入是数据湖的核心能力，通过'事务日志+ 乐观并发控制'实现ACID。Delta Lake、Iceberg、Hudi都支持并发写入，但实现细节不同。避免并发冲突的核心是：不要让多个任务写入同一分区、用MERGE INTO代替INSERT OVERWRITE、监控冲突重试次数。面试时能说清楚'乐观并发控制的原理、ACID怎么实现、实战踩坑经验'，就很稳。

下一篇聊一个很多人纠结的问题：分区分桶选错了，查询很慢？——分区策略和动态分区怎么设计？

你们生产环境遇到过并发冲突吗？怎么解决的？评论区聊聊～

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