Python中如何正确捕获org.apache.spark.sql.delta.ConcurrentAppendException异常

问题根因 #

你当前的捕获逻辑失效有两个核心原因：

• 代码语法错误：Python异常基类为大写开头的Exception，你写的小写exception属于未定义对象，捕获逻辑直接失效
• 异常封装问题：ConcurrentAppendException是JVM侧Delta内核抛出的异常，在PySpark运行环境中会被包装为Py4JJavaError，直接捕获Python侧常规异常无法命中

可行解决方案 #

1. 修正异常捕获与重试逻辑 #

首先需要导入Py4J的JavaError类，从异常堆栈中匹配Delta并发异常，示例代码如下：

from py4j.protocol import Py4JJavaError
import time
import random

max_retries = 3
retry = 0

while retry < max_retries:
    try:
        curatedTable.alias("staged").merge(
            updateDF.alias("curated"),
            "staged.ExperienceId = curated.ExperienceId AND staged.ExperienceVersion = curated.ExperienceVersion"
        ).whenMatchedUpdate(set = {"staged.updated" : "True"}).execute()
        break
    except Py4JJavaError as e:
        # 校验是否为Delta并发追加异常
        if "org.apache.spark.sql.delta.ConcurrentAppendException" in str(e.java_exception):
            retry += 1
            if retry >= max_retries:
                raise Exception("RETRY FAILED")
            # 重试前加1-3秒随机延迟，降低再次冲突概率
            time.sleep(random.uniform(1,3))
        else:
            # 非并发异常直接抛出
            raise

2. 调整Delta配置降低冲突概率 #

• 开启Delta内置提交重试：调整参数spark.databricks.delta.maxCommitAttempts = 5（默认值为3），Delta内核会自动对可重试的并发冲突进行重试，无需自己手写重试逻辑
• 分区级并发控制：如果你的Delta表是分区表，且不同管道的更新不会操作同一个分区，可以配置spark.databricks.delta.concurrentWrite.mode = "partitioned"，仅当多个作业修改同一个分区时才会触发冲突，大幅降低冲突概率
• 开启低冲突merge优化：配置spark.databricks.delta.merge.enableLowShuffleConflictDetection = true，减少merge操作的冲突检测范围

3. 架构层面优化 #

• 给Delta表按业务字段（比如ExperienceId的前缀、业务域）设置分区或分桶，将不同管道的更新拆分到不同的分区/桶中，从根源减少冲突场景
• 若允许小延迟，可引入消息队列对更新操作做缓冲，串行化写入Delta表，彻底规避并发冲突

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