Apache Iceberg高并发写入一致性：快照隔离与乐观并发控制

刚接触 Apache Iceberg 的人，十有八九会先被它的“快照隔离”“ACID 能力”吸引，但等到真正上了高并发写入的场景，才发现一致性的核心并不只是“能不能读旧版本”，而是“多个任务同时写同一张表时，系统怎么保证数据不错、不丢、最终还能读”。我最早用 Iceberg 做实时数仓时，也被并发提交时的冲突报错坑过几轮，所以这篇文章就直接从 Iceberg 的数据一致性机制切入，把它的元数据设计、快照隔离、乐观并发控制、冲突解决策略这些核心原理拆开讲。即使你没深入用过 Iceberg，这篇文章也适合你建立一张完整的技术地图——搞清楚 Iceberg 到底是用什么方式在高并发下守住一致性的，以及实际使用时你应该躲开哪些坑。

1. 内容整体设计与思路拆解

1.1 Iceberg 为什么能在高并发下保证一致性

很多传统数仓在解决并发写入时，靠的是“锁表”“锁分区”，最多做到表级锁。但 Iceberg 不一样，它把一致性的关键放在“元数据层”而不是“数据文件本身”，这一点特别重要。因为表的数据文件数量可能成百上千，而且每天都要经历写入、合并、删除，如果每次操作都要去锁住一堆文件，并发能力立刻崩溃。

Iceberg 的设计思路是：数据文件本身是不可变（immutable）的，任何写操作都只生成新文件，不修改旧文件。而表的状态则通过一份元数据文件（Metadata File）和一组不断演进的快照（Snapshot）来记录。当多个并发任务同时提交时，它们实际上是在竞争“更新同一个表的当前元数据指针”。谁能成功地把新的元数据文件提交上去，谁的操作就算成功。

这套思路本质上就是常见的乐观并发控制（Optimistic Concurrency Control, OCC）。在大多数场景下，上游的多个写入任务互不重叠，各自写各自的文件，冲突概率很低；万一真冲突了，提交方会根据 Iceberg 的版本校验机制重新判断，要么重试，要么失败并回滚。也就是用“检测冲突 + 重试”代替了“提前锁死”，从而让高并发写入成为可能。

1.2 高并发场景下的核心诉求拆解

我们说的“高并发数据一致性”，在实际工程里至少要拆成三件事：

• 多个写任务并发提交不互相覆盖：比如两个 Spark 任务同时在同一个 Iceberg 表里插入数据，各自写了自己的数据文件。如果缺乏一致性机制，后提交的任务可能覆盖掉前一个任务的元数据信息，导致前一批数据“丢失”。
• 读任务在写入时还能看到一致快照：数据仓库场景下，大部分时间都在读历史数据。如果读任务被正在发生的写入干扰，看到一半新数据、一半旧数据，就严重违背了逻辑一致性。
• 数据合并、清理等后台任务与正常写入并发时不出乱子：Iceberg 的 Compaction、过期快照删除、孤儿文件清理都会和正常写入同时进行。如果处理不当，可能出现正在读的数据文件被清理掉，或并发执行了重复合并导致元数据错乱。

Iceberg 之所以在业界被广泛用于构建湖仓一体架构，核心就是这套“基于快照 + 原子替换元数据”的机制能同时满足以上三点。它牺牲了一点写入延迟，换取高并发下的一致性保障，这个平衡在数据湖场景中非常划算。

1.3 为什么选择“乐观并发”而不是“悲观并发”

悲观并发控制最典型的就是数据库里的行锁、表锁。Iceberg 没有全局锁，也不建议你去外部引入分布式锁来包住写入操作，原因有两个：

第一，数据湖场景的大多数写入并非高频写同一行，而是高频写新文件。比如日志数据落湖，每个任务写入不同的分区新文件，冲突概率本来就低，悲观锁反而会制造不必要的等待和死锁风险。

第二，Iceberg 底层支持多种 Catalog（Hive Metastore、Hadoop、JDBC、REST 等）和文件系统。如果依赖悲观锁，就必须要求所有引擎都走同一个锁服务，这会严重削弱 Iceberg 的多引擎能力。而乐观并发控制只需要底层提供“原子的比较并交换”（CAS）能力，这个在 Hive Metastore 中通常借助事务性 SQL 来实现，在对象存储中则借助条件更新（如 ETag 或 version-id）来实现，兼容性好很多。

我自己实际用下来的体会是：Iceberg 把并发控制从“操作数据文件”转移到了“更新元数据指针”，这种设计极大地解放了高并发写入的吞吐量，同时又把一致性收敛到了一个非常薄、非常可控的边界上。

2. 核心细节解析与实操要点

2.1 元数据三层结构：Metadata File、Manifest、Data File

要理解 Iceberg 的一致性，必须先把它的元数据三层结构记熟。Iceberg 把表的所有状态组织为三个层级：

这套设计最大的好处是：描述数万个小文件的状态，只需要几个 manifest 文件；而每次提交只需要生成一个新的 metadata file 和一个 manifest list，元数据量非常小，因此“原子地替换当前元数据指针”就成为一个很轻的操作。

2.2 快照 Snapshot：把表状态变成不可变版本

快照是 Iceberg 一致性的灵魂。你可以把快照理解成“表在某个时刻的完整视图”：它对应一个 manifest list，这个 list 又串联了所有属于该快照的 manifest 和数据文件。Iceberg 的核心亮点在于，每个快照都是一旦生成就不可变的，后续任何写操作都不修改已有快照，而是追加新快照。

举个例子：凌晨 1 点有一个任务往表里写入了新分区数据，产生了快照 S1；紧接着凌晨 1 点 5 分，另一个任务又提交了新一轮数据，产生快照 S2。此时表的当前快照是 S2，但 S1 依然存在，只是不再作为“当前快照”，它代表的是“凌晨 1 点那一刻”的完整数据视图。

如果某个读任务在凌晨 1 点开始执行，它拿到的时间戳在 S1 和 S2 之间，那么它会读取 S1 对应的快照，绝不可能读到 S2 新提交的数据，也不会漏掉 S1 里已提交的数据。这就是快照隔离（Snapshot Isolation）的基本含义：读操作不使用全局锁，也能看到一致的、完整的历史版本。

2.3 原子提交：一致性关键中的关键

如果要我说出 Iceberg 唯一不能妥协的底层要求，那一定是“元数据文件的原子替换”。每个提交操作的本质流程如下：

1. 根据当前最新的元数据文件，在内存里做数据写入的规划（比如生成新的数据文件路径）。
2. 完成任务所需的实际数据文件写入。
3. 生成新的 manifest、manifest list 和 metadata file，新 metadata file 中“当前快照”指向新生成的快照。
4. 将新 metadata file 原子地“覆盖”旧 metadata file，使表的当前状态切换到新版本。

第 4 步是所有一致性保证的命门。Iceberg 要求底层 Catalog 对这个替换操作提供原子性，也就是说，多个任务同时尝试替换时，最终只能有一个成功，其余必须明确失败，然后由上层决定重试还是报错。

在 Hive Metastore 中，Iceberg 通过一个带版本字段的表来记录当前元数据位置。提交时使用类似UPDATE ... WHERE version = 旧版本的 SQL 操作，通过数据库行锁和版本号实现 CAS；在 AWS Glue Catalog 或 REST Catalog 中，通常会利用服务端版本控制实现类似效果；在 Hadoop Catalog 下则依赖文件系统 rename 的原子性，但有风险，所以现在生产上推荐最少用 Hive、JDBC 或 REST Catalog。

2.4 表版本校验与乐观锁

Iceberg 在实现并发冲突检测时，并不仅仅依赖底层 Catalog 的行锁，它自己还维护了一套版本校验逻辑。在更新表属性、提交快照等操作中，会携带“预期的元数据位置”或“基础版本号”。如果当前元数据位置已经变了，提交就会失败，并抛出类似CommitFailedException的异常。这个异常就是提醒你：提交前基于的版本已经过期，不能盲目覆盖。

不过我在实际排查中也见过不少人把“版本冲突”和“数据错误”混为一谈。版本冲突恰恰说明一致性机制在起作用——它在保护你的表不被旧任务覆盖。真正的风险是案例里忽略了冲突检测，强制用SET TBLPROPERTIES('format-version'='2')或绕过 Iceberg API 直接改元数据，那才可能导致数据损坏。

2.5 并发写的重要前置条件：显式排序和分区裁剪

尽管 Iceberg 支持并发写，但它不会替你解决所有“业务顺序”问题。例如，两个任务同时往同一个分区写数据，且业务上要求其中一个必须覆盖另一个，这时候需要你在应用层指定顺序或使用MERGE INTO之类的操作。Iceberg 的一致性指的是“系统层面不会产生脏写、丢数据”，但“业务层面谁覆盖谁”需要你自己定义好。

这里我总结几个常用的实操策略：

• 写任务尽量使用确定性分区键，让不同任务写入不同文件，降低冲突概率。
• 如果两个任务可能修改同一批数据，尽量给它们指定顺序，不要同时提交。
• 合理设置提交重试次数和间隔，因为偶发的提交冲突在 Iceberg 中是正常的，不一定需要人工介入。

3. 实操过程与核心环节实现

3.1 从建表到并发提交：一个标准的 ETL 流程

我用一段简单的 SQL 来演示，先建一个分区表，然后用并发任务写入数据。

CREATE TABLE lake.events ( event_id BIGINT, user_id BIGINT, event_type STRING, event_time TIMESTAMP ) USING iceberg PARTITIONED BY (days(event_time));

假设我们有两个 Spark 批处理任务，分别处理某天的两批日志：

// 任务A spark.sql("INSERT INTO lake.events SELECT /*+ SHUFFLE_PARTITIONS(10) */ * FROM raw_logs WHERE day = '2024-06-01'")

// 任务B spark.sql("INSERT INTO lake.events SELECT * FROM other_source WHERE day = '2024-06-01'")

这两个任务如果同时提交，Iceberg 会如何处理？任务 A 和任务 B 各自写入自己的数据文件，然后在提交阶段都去尝试更新表的当前元数据。假设任务 A 先提交成功，表的当前快照切换到 A；任务 B 提交时发现“当前元数据已不是自己读取时的版本”，提交失败。

这个失败并不是系统错误，而是预期的冲突。Iceberg 客户端通常会自动重试：任务 B 重新读取新的当前元数据，重新规划写入（此时会带上任务 A 已写入的文件信息），然后再次提交。在 Iceberg 的写法中，这个过程就是所谓的表刷新（refresh table）和重试提交。

3.2 使用 Flink / Spark 时的并发提交参数

不同的计算引擎对 Iceberg 的支持方式不太一样，但底层都遵循同一套提交协议。

在 Spark 3.x + Iceberg 的场景下，你需要在 SparkSession 里设置：

spark.conf.set("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") spark.conf.set("spark.sql.catalog.demo", "org.apache.iceberg.spark.SparkCatalog") spark.conf.set("spark.sql.catalog.demo.type", "hive") spark.conf.set("spark.sql.catalog.demo.uri", "thrift://metastore-host:9083")

在写任务较多时，可以调低文件生成大小，让写入的数据文件更小、更碎，减少单个文件上的资源竞争，但代价是后续查询性能下降，需要靠 Compaction 来平衡。我个人建议保持默认的 128MB 或 256MB 文件大小，只有遇到明显的提交超时才考虑调小。

Flink 里则主要配置：

IcebergSink.builder(flinkEnv) .table(icebergTable) .overwrite(false) .build()

Flink 的流式写入是持续提交小批次的，默认开启的是两阶段提交协议（Two-Phase Commit，在 Flink 里叫 Exactly-Once 或写提交算子的 checkpoint 时提交）。每个 checkpoint 对应一次 Iceberg 提交，所以并发度需要控制好，否则容易产生大量碎片文件。

3.3 实时写入高并发下的文件合并策略

很多人以为 Iceberg 只要天然支持 ACID，就可以不管小文件数量了，这其实是大忌。高并发实时写入最容易产生海量的小文件，每个 Flink checkpoint 都可能生成一批新文件，长年累月下来，元数据膨胀得厉害，查询时打开的文件数成百上千，性能断崖下跌。

为了平衡一致性和性能，我常用的策略有几类：

• 写后自动合并（Auto Compaction）：在 Spark 写入后触发REWRITE_DATA_FILES，但注意要和正在进行的写入任务做好并发控制，避免合并时和新的写入冲突。
• Flink 写入端开启文件合并：有的版本支持在 sink 端按 checkpoint 生成的文件做合并，可以减少小文件数量。
• 定期离线 Compaction：使用ALTER TABLE lake.events EXECUTE rewrite_data_files这种操作，选在低峰期执行。

ALTER TABLE lake.events EXECUTE rewrite_data_files( using = 'sort', options = Map('max-file-group-size-bytes' -> '268435456') );

关键一点：Compaction 本身也是提交操作，它也会走乐观并发流程。如果你在 Compaction 过程中有新的写入产生，二者可能冲突。Iceberg 的解决方案是让 Compaction 在“不删除原文件，只生成新文件”的基础上提交，并通过在元数据里记录文件组信息来避免重复合并。生产上我会在 Compaction 任务上做一些限流和调度，尽量避免与凌晨高峰写入重叠。

3.4 通过 SQL 观察快照演进

看明白 Iceberg 的一致性，最直观的方式是直接查看表的快照历史：

SELECT * FROM lake.events.snapshots ORDER BY committed_at;

+-------------------------+----------------+---------------------+-------------------+ | snapshot_id | parent_id | committed_at | operation | +-------------------------+----------------+---------------------+-------------------+ | 6244271450482345678 | null | 2024-06-01 01:00:00 | append | | 6244271450482345679 | 624427145048...| 2024-06-01 01:05:00 | append | | 6244271450482345680 | 624427145048...| 2024-06-01 03:00:00 | overwrite | +-------------------------+----------------+---------------------+-------------------+

每个快照都有一个父快照 ID，通过它可以串联出完整的版本链。当出现数据异常时，利用time travel可以直接查某个历史快照的数据，比如：

SELECT * FROM lake.events VERSION AS OF 6244271450482345679 LIMIT 10;

这在排查并发写入造成的数据异常时极其有用。你不需要恢复整个表，只需要针对某个怀疑的快照做验证。

4. 常见问题与排查技巧实录

4.1 提交冲突：CommitFailedException

这应该是你使用 Iceberg 高并发写入时最常遇到的异常。报错信息通常类似：

CommitFailedException: Cannot commit: The table has been updated by another process after this process has started.

很多人第一次见到会慌，其实解决方法很简单。

我的建议是：

• 确认你的写入是否真的有必要和另一个任务“同时抢跑”。如果业务上允许串行，那就用调度器把任务错开。
• 如果是流式写入，开启 Iceberg 的重试机制。乐观并发控制本身就是撞上了再重试，不是一开始就规避。
• 观察冲突频率，如果每天冲突几十次，说明可能是同一张表的写入任务太多太碎，考虑合并写入任务或在做轻量写合并后再提交。

此外，很多 Iceberg 的 API 里提供了retry参数，例如 Spark 的MERGE INTO操作可以设置：

MERGE INTO lake.events t USING updates u ON t.event_id = u.event_id WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *

Spark 端会基于提交失败自动重试几次，默认重试次数是 4 次，可以在 Catalog 属性里调大。不过重试次数不是越多越好，因为每次重试都要重新读取一次元数据和规划任务，重试太频繁会导致提交耗时成倍增长。

4.2 读任务看到的数据“不新”是怎么回事

有的开发者会遇到这种问题：数据明明写进去了，但下游 SELECT 看不到新数据。这很可能不是因为写入没成功，而是因为读任务在读一个“旧的快照”。

检查方法很简单：

SELECT snapshot_id FROM lake.events.snapshots ORDER BY committed_at DESC LIMIT 1;

如果读任务使用的快照 ID 不是这个最新的 ID，说明你的读取端拿到了旧快照。常见原因有：

• 读任务是长事务，启动时快照已确定，事务期间即使有新的提交也不会感知。
• 通过某些缓存层读取数据，缓存了旧的元数据位置。
• 使用时间旅行读取了某个历史时间点的快照，但业务上误以为在读最新。

解决办法是确保读端在每次查询时刷新表状态，不要复用过长的 Iceberg Table 对象。尤其是 JDBC 或 Spark Thrift Server 场景下，连接复用可能导致元数据缓存滞后。

4.3 快照过多导致元数据膨胀

Iceberg 虽然支持时间旅行，但老快照不清理的话，元数据会一直膨胀。每个快照对应一组 manifest 文件，如果不定期清理，查询时 Iceberg 需要加载的 manifest 越来越多，性能会下降。

我建议设置合理的表属性：

ALTER TABLE lake.events SET TBLPROPERTIES ( 'write.metadata.delete-after-commit.enabled' = 'true', 'write.metadata.previous-versions-max' = '50' );

这会自动保留最近 50 个版本，超出部分在每次提交时异步清理。这样可以同时兼顾数据恢复和历史清理。对于审计要求保留 30 天数据的场景，要把previous-versions-max适当调大，或者每天定时执行过期快照清理。

4.4 常见问题速查表

4.5 关于“乐观并发”的边界条件和避坑经验

最后聊几个我自己实践过后才理解的边界条件。

Iceberg 的乐观并发依赖“比较并交换”的原子性，但它并不能保证用户自定义的“业务逻辑”在多任务下正确。比如，某个 CDC 场景中，有两个任务分别读取同一批 binlog 数据，都执行MERGE INTO，如果任务没有按主键做去重和顺序控制，后提交的任务可能覆盖前一个任务的更新。Iceberg 只保证提交本身是原子的，但不负责业务层面的幂等。这个风险需要在上游做处理，比如使用 Flink CDC 的 upsert 模式，或者在 ETL 层对日志做去重。

再有一点，Iceberg 表的多写入任务如果同时做DELETE操作，冲突概率会比INSERT高很多。因为 DELETE 需要根据匹配条件读取整表数据，并在提交时删除多个数据文件。如果你有两个删除任务同时执行，二者基于的文件列表很可能重叠，提交时就很容易冲突。此时可以考虑把删除操作也做成“指定分区 + 指定条件”的精确操作，而不是全表扫描式删除。

还有一点要特别提醒：不要在提交失败后直接“忽略异常”。有些开发者为了减少任务失败率，在捕捉到 CommitFailedException 后就静默处理，这会导致一批数据悄悄丢失。正确做法是至少打报警，并让任务进入等待重试或人工介入流程。

4.6 多核高并发场景下的提交优化

借助当前多核处理器的并行能力，Iceberg 的写入端经常会在同一节点上开很多并发 writer。比如 Spark Executor 上每个 Task 写一个数据文件，几十个 Task 同时提交到同一个表时，即便任务之间没有业务冲突，也可能因为共享 Catalog 的 CAS 操作而出现一定概率的提交竞争。

针对这种“伪冲突”，我的优化手段是：

• 把写入表分区设计得更细，例如按天 + 小时分区，减少同一时刻多个任务写同一分区的概率。
• 写入的数据文件数不要无限增加：单任务内使用fanout写入时，控制文件数量在一个合理范围。
• 如果多个任务写同一张表，可以给每个任务指定不同的write.order，让文件的生成位置尽量分散。
• 调整 Catalog 端的连接池大小，避免提交时产生连接瓶颈。

这些优化做下来，提交冲突率能下降很多，整体写入吞吐量也随之提升。

5. 结尾

我做了几年数据架构，用过的表格式不少，Iceberg 在一致性上的设计确实让我觉得“聪明”：它没有把所有并发控制都压在锁服务上，而是用不可变快照、原子元数据替换、乐观并发检测，把一致性收敛到一条很清晰的链路上。这种设计思路对高并发写入非常友好，也给了上层计算引擎很大的调度自由。

如果你第一次上手遇到 CommitFailedException，别慌，那正是 Iceberg 的守卫机制在发挥作用。你要做的是理解冲突来源，然后把任务适当错峰、开启重试、合理保留快照、定期 Compaction。真正麻烦的从来不是“冲突报错”，而是冲突发生时你没有任何感知，那才叫数据事故。最后再分享一个我自己常备的检查习惯：每次上完批量任务，我都会查一下这条 SQL——

SELECT operation, summary, committed_at FROM lake.events.snapshots ORDER BY committed_at DESC LIMIT 20;

它能快速告诉你最近表经历了哪些提交操作，是 append 还是 overwrite，是不是全部成功。数据一致性的本质，很多时候不在于一次完美的设计，而在于你能不能在出问题的最初几分钟里，准确地定位到版本、快照和提交记录的异常。掌握 Iceberg 这套机制，就是拿到了洞察湖仓数据一致性的那把钥匙。