PySpark Append模式SaveAsTable并发写入行为及报错解决咨询

问题背景 #

尝试以Append模式向表写入数据，使用代码如下：

df.repartition(5).write.mode("append").partitionBy("partition_id").option("path", path).saveAsTable(table)

需明确两个核心问题：

1. PySpark是否推荐对表进行并发写入？
2. 如何解决测试场景中遇到的错误？

测试场景与现象 #

场景1：2个并发作业

• 情况1：表不存在（首次创建）
  • 现象：一个作业成功，另一个因Table already exists错误失败
• 情况2：表已存在
  • 现象：两个作业均成功，无报错

场景2：4个并发作业

• 情况1：表不存在（首次创建）
  • 预期：与场景1情况1行为一致
• 情况2：表已存在
  • 现象：1个作业成功，其余3个首次失败，重试后成功，错误信息如下：
    • 失败作业1错误：

      WARN org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter: Exception get thrown in job commit, retry (1) time.
      java.io.IOException: Upload failed for '<base_path>/_SUCCESS'
      

      根源错误：

      Caused by: com.google.cloud.hadoop.repackaged.gcs.com.google.api.client.googleapis.json.GoogleJsonResponseException: 412 Precondition Failed
      PUT https://storage.googleapis.com/upload/storage/v1/b/<bucket>/o?ifGenerationMatch=<timestamp_may_be>&name=<base_path>/_SUCCESS&uploadType=resumable&upload_id=<some_token>
      {
        "code" : 412,
        "errors" : [ {
          "domain" : "global",
          "location" : "If-Match",
          "locationType" : "header",
          "message" : "At least one of the pre-conditions you specified did not hold.",
          "reason" : "conditionNotMet"
        } ],
        "message" : "At least one of the pre-conditions you specified did not hold."
      }
      

    • 失败作业2错误：

      Caused by: java.io.IOException: Failed to merge directory from <base_path>/_temporary/0/_temporary/attempt_20230415171143_0002_m_000004_41/partition_id=1234 to <base_path>/partition_id=1234
      
      Caused by: java.util.concurrent.ExecutionException: java.io.IOException: Failed to rename FileStatus{path=<base_path>/_temporary/0/_temporary/attempt_20230415171143_0002_m_000004_41/partition_id=1234/part-00004-91590a94-c0ba-4ab6-bb49-bfeea928d6e8.c000.snappy.parquet; isDirectory=false; length=848073403; replication=3; blocksize=134217728; modification_time=1681578746177; access_time=1681578746177; owner=<username>; group=<username>; permission=rwxrwxrwx; isSymlink=false} to <base_path>/partition_id=1234/part-00004-91590a94-c0ba-4ab6-bb49-bfeea928d6e8.c000.snappy.parquet
      

    • 失败作业3存在类似错误

Spark版本：2.4.8

问题解答 #

1. PySpark是否推荐并发Append写入同一张表？ #

PySpark支持并发Append写入，但需结合存储层和业务场景调整：

• 若并发作业写入的分区不重叠，冲突概率极低，是推荐的高效写入方式；
• 若存在分区重叠，易出现文件重命名、元数据竞争等问题，需通过配置或业务逻辑规避。

2. 针对测试场景错误的解决方案 #

（1）表首次创建时的Table already exists错误

• 原因：多个作业同时尝试创建表，元数据操作（如Hive Metastore表创建）是原子性的，但作业间无协调，先完成的作业创建表后，后续作业触发存在性检查错误。
• 解决方案：
  • 提前初始化表：在所有并发作业启动前，通过空DataFrame写入或CREATE TABLE语句预定义表结构，避免作业竞争创建；
  • 增加重试逻辑：捕获Table already exists异常，短暂等待后重试，此时表大概率已被其他作业创建完成，可正常执行Append。

（2）表已存在时多作业并发的报错（412错误、文件重命名失败）

这类错误核心是多作业提交阶段竞争同一资源（如_SUCCESS文件、同一分区目录），尤其在GCS等对象存储上，文件操作原子性与HDFS存在差异，更易触发冲突。

针对_SUCCESS文件的412错误

• 原因：每个作业完成后都会尝试写入_SUCCESS文件，多作业同时操作时，GCS的预条件检查会因文件已被修改返回412错误。
• 解决方案：
  • 禁用_SUCCESS文件生成：添加配置spark.hadoop.mapreduce.fileoutputcommitter.marksuccessfuljobs=false，避免该资源的竞争；
  • 升级Spark版本：Spark 3.x针对对象存储的提交逻辑做了优化，可大幅减少此类冲突；若无法升级，可调整提交器配置适配存储特性。

针对分区目录文件重命名失败的错误

• 原因：多作业同时向同一partition_id分区写入，提交阶段移动临时文件到正式目录时，出现文件名重复或目录操作冲突。
• 解决方案：
  • 按分区分片写入：业务允许的话，将作业按partition_id拆分，每个作业仅负责特定分区，避免同一分区的并发写入；
  • 调整提交器配置：设置spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2（默认是1），版本2的提交器会先将文件写入临时目录，最后一次性移动，降低冲突概率；注意需验证对象存储兼容性；
  • 增加作业重试机制：这类冲突属于瞬时性问题，捕获IO异常后重试作业即可成功（如测试中重试后成功的现象）；
  • 使用存储适配的提交器：针对GCS，使用com.google.cloud.hadoop.io.gcs.GoogleHadoopMapReduceCommitter，更好适配对象存储特性，减少冲突。

额外建议 #

• 控制并发作业数量：通过队列调度（如YARN队列）限制同时写入的作业数，降低冲突概率；
• 监控重试次数：避免无限重试，设置合理的重试阈值，同时记录冲突日志用于后续优化。

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