Spark SQL含聚合/日期函数的DataFrame写入HDFS报错求助

看起来你遇到的问题是Spark SQL查询能正常预览结果，但写入HDFS时失败——这是个很常见的坑，核心原因在于show()和写入操作对Spark执行计划的触发程度不同：show()默认只计算前20条数据，不会触发完整的shuffle和全量聚合；而写入HDFS需要处理所有数据，会暴露show()阶段没触发的潜在问题。结合你使用了聚合函数、日期函数和全外连接，以下是最可能的原因和解决办法：

1. 数据倾斜导致全量计算失败 #

如果你的分组字段year(plt_date)/month(plt_date)存在某个时间区间数据量极大（比如某一个月的数据占了总数据的70%以上），show()只会取前20条数据，不会触发完整的shuffle；但写入时需要把所有数据按年月分组，会导致对应分区的数据量远超其他分区，引发内存溢出（OOM）或任务超时。

解决办法： #

• 启用Spark自适应执行与倾斜优化：在创建SparkSession时添加配置，让Spark自动处理倾斜：

  spark = (SparkSession.builder
           .appName("appName")
           .enableHiveSupport()
           .config("spark.sql.adaptive.enabled", "true")
           .config("spark.sql.adaptive.skewJoin.enabled", "true")
           .config("spark.sql.adaptive.skewJoin.skewedPartitionThreshold", "100000") # 可根据你的数据量调整阈值
           .getOrCreate())
  

• 手动盐值打散分组键（倾斜严重时使用）：给分组键添加随机前缀先做局部聚合，再合并结果：

  SELECT Year, Mounth, SUM(B_Count) AS B_Count, SUM(P_Count) AS P_Count
  FROM (
      SELECT 
          year(plt_date) AS Year, 
          month(plt_date) AS Mounth, 
          count(build) AS B_Count, 
          count(product) AS P_Count,
          CAST(RAND() * 10 AS INT) AS salt -- 将分组打散为10个临时分组
      FROM first_table 
      FULL OUTER JOIN second_table ON key1=CONCAT('SS',key_2) 
      GROUP BY year(plt_date), month(plt_date), salt
  ) t
  GROUP BY Year, Mounth
  

2. 空值引发的聚合/写入异常 #

全外连接会产生大量null值：如果plt_date本身是null，year(null)/month(null)会返回null，所有null值会被分到同一个分区，可能导致该分区数据量爆炸；另外部分存储格式（比如早期的TextFile）对null的兼容性较差。

解决办法： #

• 过滤或处理空日期（业务允许的情况下）：

  SELECT 
      year(plt_date) AS Year, 
      month(plt_date) AS Mounth, 
      count(build) AS B_Count, 
      count(product) AS P_Count 
  FROM first_table 
  FULL OUTER JOIN second_table ON key1=CONCAT('SS',key_2) 
  WHERE plt_date IS NOT NULL -- 过滤无日期的行
  GROUP BY year(plt_date), month(plt_date)
  

• 使用兼容空值的存储格式：推荐用Parquet或ORC格式写入，它们对null的支持更完善：

  result_df = spark.sql("你的完整查询语句")
  result_df.write.mode("overwrite").parquet("hdfs://your/target/path")
  

3. 写入路径与模式问题 #

如果之前的写入操作失败，HDFS路径可能已经存在，Spark默认的写入模式是errorifexists，会直接报错；而简化版本数据量小，可能在测试时覆盖路径没触发问题。

解决办法： #

• 明确指定写入模式，比如覆盖已有路径：

  result_df.write.mode("overwrite").format("parquet").save("hdfs://your/target/path")
  

• 若需要追加数据，使用append模式：

  result_df.write.mode("append").format("parquet").save("hdfs://your/target/path")
  

4. 数据类型兼容性问题 #

count()函数返回的是BigInt类型，部分存储格式（比如某些自定义格式）对大整数的映射可能存在问题，虽然show()能正常显示，但写入时会报错。

解决办法： #

• 显式转换字段类型：

  SELECT 
      year(plt_date) AS Year, 
      month(plt_date) AS Mounth, 
      CAST(count(build) AS INT) AS B_Count, 
      CAST(count(product) AS INT) AS P_Count 
  FROM first_table 
  FULL OUTER JOIN second_table ON key1=CONCAT('SS',key_2) 
  GROUP BY year(plt_date), month(plt_date)
  

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