Spark Streaming 2.4.0输出Parquet缺失uncompressed_page_size字段问题咨询

这个报错本质是你生成的Parquet文件元数据不完整，缺少了Parquet格式规范要求的uncompressed_page_size字段，在Spark 2.4.0版本里，主要有这几个常见原因：

核心原因分析 #

• 小文件/极小数据量写入bug：Spark 2.4.x的Parquet Vectorized Writer在处理极小数据集（比如只有几行甚至一行数据）时，会出现元数据写入不完整的问题。当数据量小到不足以填满一个Parquet Page时，Writer可能会跳过部分必填元字段的写入，直接导致文件损坏。
• 写入过程中断：如果你的Streaming任务出现Executor意外退出、任务重试或者批次超时，Parquet写入的临时文件可能无法正常重命名为最终文件，这些残留的不完整文件会被parquet-tools或Athena当成有效文件读取，从而触发报错。
• 碎片化文件过多：你的代码每次批次直接写入，没有做文件合并处理，导致大量小Parquet文件生成，每个小文件都有更高的概率出现元数据写入异常。

针对性解决方案 #

1. 优先升级Spark版本（最彻底） #

Spark 2.4.0确实存在不少Parquet相关的已知bug，后续的2.4.5+版本以及3.x系列已经修复了这类元数据缺失的问题。如果业务允许，升级到2.4.5或更高版本是最省心的解决办法。

2. 优化写入逻辑，减少小文件 #

修改你的写入代码，添加文件合并逻辑，同时确保写入的原子性：

// 建议在Streaming应用启动时就初始化SparkSession，避免在foreachRDD中重复创建
val spark = SparkSession.builder()
  .config(rdd.sparkContext.getConf)
  .appName("YourStreamingApp")
  .getOrCreate()
import spark.implicits._

ds.foreachRDD { rdd =>
  val filteredRDD = rdd.filter(_.isDefined).map(_.get)
  if (!filteredRDD.isEmpty()) {
    val now = utcNow()
    val location = s"${appConfig.output}/${datef(now, "yyyyMM")}/${datef(now, "yyyyMMdd")}/${datef(now, "yyyyMMddHH")}/${appConfig.exchange}${datef(now, "yyyyMMddHHmmss")}"
    filteredRDD.toDF()
      .repartition(1) // 每个批次生成一个文件，减少碎片化
      .write
      .mode(SaveMode.Append) // 允许同一目录下多次写入（应对批次重试）
      .parquet(location)
  }
}

这样修改的好处：

• 提前初始化SparkSession，避免重复创建的开销和上下文冲突
• repartition(1)将每个批次的数据合并为一个文件，降低小文件带来的元数据异常概率
• 使用Append模式，避免批次重试时因文件已存在而报错，同时保证写入的原子性

3. 临时规避：关闭Vectorized Writer #

如果暂时无法升级Spark，可以尝试关闭Parquet的Vectorized Writer（虽然会牺牲一点性能，但能规避小文件的元数据bug）：

filteredRDD.toDF()
  .write
  .option("spark.sql.parquet.enableVectorizedWriter", "false")
  .parquet(location)

4. 事后补救：清理损坏文件 #

对于已经生成的损坏文件，可以用parquet-tools meta <file-path>命令批量校验，找出损坏的文件并删除，避免Athena读取时报错。

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