Python Polars写入Delta Lake并发异常求助

使用Python Polars将DataFrame追加写入本地Delta Lake，当同时运行多个脚本实例时，先启动的脚本会无报错终止。已采用append模式，并按source、year、month、day、hour字段分区，预期该配置能隔离写入、缓解冲突，但问题仍出现。相关代码如下：

df.write_delta(
    "data/collect",
    mode="append",
    delta_write_options={
        "partition_by": ["source", "year", "month", "day", "hour"],
    },
    storage_options= {
        "compression": "ZSTD",
        "compression_level": "22",
    },
)

原因分析 #

即使使用分区和追加模式，Delta Lake的写入操作仍需修改表的元数据（事务日志），本地文件系统的并发写入会触发乐观并发控制冲突：

1. 多个实例同时尝试更新元数据时，后完成提交的实例会覆盖先启动的实例的元数据变更，导致先启动的写入操作被静默终止（部分Delta/Polars实现未抛出异常）。
2. 若不同实例写入同一个分区路径，哪怕是追加模式，也会因文件写入冲突触发相同问题。

解决方案 #

1. 确保分区完全隔离 #

验证每个脚本实例的DataFrame中，source、year、month、day、hour的组合完全唯一，避免多个实例写入同一个分区目录。如果存在分区重叠，冲突无法通过配置避免。

2. 配置Delta冲突处理策略 #

在delta_write_options中添加writeConflictMode参数，指定宽松的冲突处理规则，允许并发追加：

df.write_delta(
    "data/collect",
    mode="append",
    delta_write_options={
        "partition_by": ["source", "year", "month", "day", "hour"],
        "writeConflictMode": "append"  # 允许并发追加，仅在数据真正冲突时终止
    },
    storage_options= {
        "compression": "ZSTD",
        "compression_level": "22",
    },
)

writeConflictMode的可选值：

• abort（默认）：只要检测到元数据冲突就终止写入
• append：仅当写入的数据与现有数据存在逻辑冲突（如主键重复）时终止，允许无冲突的并发追加
• ignore：静默忽略冲突，直接丢弃冲突的写入操作（不推荐）

3. 升级依赖版本 #

确保使用最新版本的Polars和delta-rs（Polars的Delta写入依赖该库），旧版本可能存在本地文件系统并发写入的bug：

pip install --upgrade polars delta-rs

4. 启用本地文件锁 #

对于本地Delta Lake，可通过delta_write_options指定本地锁提供商，增强并发写入的可靠性：

df.write_delta(
    "data/collect",
    mode="append",
    delta_write_options={
        "partition_by": ["source", "year", "month", "day", "hour"],
        "writeConflictMode": "append",
        "lockProvider": "io.delta.storage.LocalLockProvider"
    },
    storage_options= {
        "compression": "ZSTD",
        "compression_level": "22",
    },
)

内容的提问来源于stack exchange，提问作者Scott Syms