在多作业并发写入或流批混写同一张 Paimon 表时，Paimon 是如何检测并解决写入冲突的？

知识点图片

根据 Apache Paimon（Master 分支）的官方文档，当存在多作业并发写入或流批混写同一张 Paimon 表时，Paimon 主要通过乐观并发控制（Optimistic Concurrency Control, OCC）机制来检测并解决写入冲突。

以下是 Paimon 检测和解决写入冲突的具体原理、冲突类型以及推荐的最佳实践：

一、核心机制：乐观并发控制（OCC）

在 Paimon 中，每个写入作业都按照自己的节奏写入数据。当作业尝试提交（Commit）时，它会基于当前最新的 Snapshot（快照），应用增量文件（新增数据文件或逻辑删除旧文件）来尝试生成一个新的 Snapshot。

在此过程中，Paimon 主要检测并处理以下两类冲突：

1. 快照冲突 (Snapshot Conflict)

冲突原因：当一个作业准备提交其生成的 Snapshot 时，发现它计划使用的 snapshot-id 已经被其他并发作业抢占并提交了（即表已经生成了由其他作业产生的最新 Snapshot）。
解决办法：
- 在支持原子 RENAME 的文件系统上（如 HDFS）：Paimon 利用文件系统的原子重命名（Atomic Rename）机制来提交 Snapshot。检测到 ID 被抢占后，当前作业会自动重试提交（获取最新的 snapshot-id 并重新尝试提交），从而保证事务的 ACID 特性。
- 在不支持原子 RENAME 的对象存储上（如 S3、OSS 等）：对象存储的 RENAME 操作不具备原子语义，这可能导致快照丢失或冲突失效。为此，Paimon 引入了外部锁（Catalog Lock）。用户需要配置 Hive Metastore 或 JDBC Metastore，并开启 lock.enabled 参数（支持 zookeeper 或 hive 等锁类型），利用外部锁来协调并发提交，避免快照冲突。

2. 文件冲突 (Files Conflict)

冲突原因：当作业尝试提交文件的删除（通常是 LSM 树的 Compaction 引起的逻辑删除）时，Paimon 会对比最新的 Snapshot。如果发现当前作业想要删除的某个旧数据文件，已经被另一个作业逻辑删除并提交了，此时就会触发文件冲突。
解决办法：
- 一旦发生文件冲突，作业将无法直接在该节点继续提交，而是会主动触发一次 Failover（故障转移/重启）。
- 重启后，作业会从文件系统中重新拉取最新的表状态，重新计算并尝试解决冲突。Paimon 的这种设计可以确保数据不会丢失或重复，但如果多个流作业并发向同一分区写入并同时触发 Compaction，可能会导致作业频繁重启。

二、多作业并发/流批混写的冲突解决与优化实践

为了在使用多作业并发写入或流批混写时彻底避免上述冲突，Paimon 提供了以下最佳实践：

1. 写入不同分区（推荐）

Paimon 默认支持对不同分区进行并发写入，并且这种写入是完全隔离且安全的。

推荐模式：让流式作业（Streaming Job）持续写入 Paimon 的最新分区（如当前天/当前小时），同时让批处理作业（Batch Job，例如通过 INSERT OVERWRITE）写入历史分区。这种模式下，作业之间不会产生数据或文件冲突。

2. 独立 Compaction 任务（Dedicated Compaction Job）

如果必须有多个作业同时向同一个分区写入数据，冲突的本质通常源于作业内部自动触发的 Compaction（涉及删除旧文件和合并新文件）。为了避免并发写入时的文件冲突，Paimon 建议将 Compaction 剥离：

第一步：禁用写入作业的 Compaction
在写入表的属性中设置 'write-only' = 'true'。此时，所有的并发写入作业（流或批）将只负责写数据，不再执行 Compaction 和 Snapshot 过期清理，因此写入作业之间永远不会发生文件删除冲突。
第二步：运行一个独立的 Compaction 任务
单独启动一个作业（如 Flink 批作业或调用系统存储过程 sys.compact）来专门对该表进行后台 Compaction 合并。因为只有一个专用任务在做文件合并与删除，从而彻底消除了文件冲突导致的作业重启问题。

3. 配置 Catalog 锁

如前文所述，在多作业并发写入、且底层存储使用的是 S3/OSS 等对象存储时，务必在 Catalog 配置中启用锁机制，例如：

sql

CREATE CATALOG my_catalog WITH (
  'type' = 'paimon',
  'metastore' = 'hive', -- 或 jdbc
  'uri' = 'thrift://<host>:<port>',
  'lock.enabled' = 'true',
  'lock.type' = 'hive' -- 或 'zookeeper'
);

播面