向Azure Data Lake特定容器写入Parquet文件时出错

从同一Azure存储账户的container1读取两个文件，经转换合并得到DataFrame df_spark，操作流程为：挂载container1 → 读取并处理数据 → 卸载container1 → 挂载container2。尝试将df_spark写入container2时触发Py4JJavaError，但存在以下异常现象：

• 同一DataFrame写入container1完全正常
• 生成随机数据写入container2无异常
• 已尝试将pandas DataFrame转为Spark DataFrame，问题依旧

写入Parquet的代码：

spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
df_spark.coalesce(1).write.option("header",True) \
        .partitionBy('ZMTART') \
        .mode("overwrite") \
        .parquet('/mnt/temp/')

报错栈片段：

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<command-3769031361803403> in <cell line: 2>()
      1 spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
----> 2 df_spark.coalesce(1).write.option("header",True) \
      3         .partitionBy('ZMTART') \
      4         .mode("overwrite") \
      5         .parquet('/mnt/temp/')

/databricks/spark/python/pyspark/instrumentation_utils.py in wrapper(*args, **kwargs)
     46             start = time.perf_counter()
     47             try:
--> 48                 res = func(*args, **kwargs)
     49                 logger.log_success(
     50                     module_name, class_name, function_name, time.perf_counter() - start, signature

/databricks/spark/python/pyspark/sql/readwriter.py in parquet(self, path, mode, partitionBy, compression)
   1138             self.partitionBy(partitionBy)
   1139         self._set_opts(compression=compression)
-> 1140         self._jwrite.parquet(path)
   1141 

1. 检查container2的挂载权限与配置 #

• 确认container2的挂载命令权限配置：存储账户密钥/服务principal必须包含Write权限，仅Read权限会导致写入失败
• 验证挂载路径有效性：执行dbutils.fs.ls('/mnt/temp')，确认能正常列出container2目录内容
• 确保卸载/挂载时序正确：通过dbutils.fs.unmount('/mnt/container1')的返回值确认container1完全卸载后，再执行container2的挂载操作

2. 排查分区列ZMTART的特殊情况 #

• 检查列值是否包含非法字符：Azure存储目录名不允许\ / : * ? " < > |等字符，Parquet分区目录由ZMTART值生成，非法字符会导致写入失败
• 临时移除分区参数测试：去掉partitionBy('ZMTART')后尝试写入，若成功则定位为分区列问题，需先清洗数据：

  from pyspark.sql.functions import regexp_replace, col
  # 替换特殊字符
  df_spark = df_spark.withColumn('ZMTART', regexp_replace('ZMTART', r'[\\/:*?"<>|]', '_'))
  # 过滤空值
  df_spark = df_spark.filter(col('ZMTART').isNotNull())
  

3. 调整写入参数与模式 #

• 移除coalesce(1)：强制合并为单个文件易引发大文件写入超时或内存溢出，建议改为repartition(n)（n根据数据量设置合理值）
• 清理多余参数：Parquet格式无需option("header",True)，该参数仅适用于CSV等文本格式，多余参数可能引发兼容性问题
• 调整写入模式：若container2为空目录，dynamic分区覆盖模式可能不生效，可先改为mode("append")测试，或提前创建基础分区目录

4. 检查DataFrame的元数据与数据完整性 #

• 对比元数据一致性：执行df_spark.printSchema()，确认写入container1和container2时的DataFrame数据类型、字段可空性完全一致
• 排查损坏数据：执行df_spark.filter(col('_corrupt_record').isNotNull()).count()，若存在损坏记录需先过滤或修复

5. 获取完整报错日志 #

Py4JJavaError的底层Java异常信息才是问题核心，可通过以下方式获取：

• 在Databricks作业界面查看Driver日志或Executor日志
• 捕获异常并打印完整栈信息：

  try:
      df_spark.write.partitionBy('ZMTART').mode("overwrite").parquet('/mnt/temp/')
  except Exception as e:
      print(str(e))
      import traceback
      traceback.print_exc()
  

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