Spark Structured Streaming中Parquet写入分区失效问题求助

我来帮你捋捋这个分区失效的问题——这种情况我之前碰到过好多次，大多是几个容易忽略的细节没处理到位，咱们一步步排查：

1. 核心前提：分区列必须保留在DataFrame中 #

你提到要写入除date外的所有字段，但这里有个关键认知：partitionBy("date")要求date必须是DataFrame中的列。Spark会自动把分区列从数据文件中剥离（只作为目录路径的一部分），不需要你手动移除它。

如果你的代码里不小心把date字段从DataFrame中删掉了（比如用select排除了它），那分区逻辑肯定会失效。

❌ 错误示例：

// 错误：移除了date列，Spark找不到分区依据
val writeDF = originalDF.select("zoneId", "deviceId", "timeSinceLast")
writeDF.writeStream.partitionBy("date")... // 要么报错，要么完全不生成分区

✅ 正确做法：
保留date列在DataFrame中，Spark会自动处理它作为分区目录，不会把它写入Parquet数据文件：

// 直接使用包含date列的原DataFrame即可
val writeDF = originalDF

2. 确认date字段的类型正确性 #

分区列的类型如果不符合预期，也会导致分区逻辑异常：

• 如果date是Timestamp类型，会生成类似date=2024-05-20 14:30:00的分区目录，这可能不是你想要的按天分区效果；
• 如果是字符串类型，要确保格式是标准的日期格式（比如yyyy-MM-dd），否则Spark无法正确识别分区规则。

建议统一转成Date类型：

import org.apache.spark.sql.functions._
val preparedDF = originalDF.withColumn("date", to_date(col("date")))

3. 检查Checkpoint目录是否复用了旧作业的元数据 #

如果你的作业之前没有设置partitionBy，或者分区逻辑不同，直接复用旧的checkpointLocation会导致Spark沿用之前的元数据，忽略新的分区配置。

解决方法：删除旧的checkpoint目录，重新启动作业。

4. 排查date字段的空值情况 #

如果date字段存在大量null值，会生成date=null的分区目录，可能你误以为分区没生效。可以先验证数据的非空情况：

// 查看date字段的非空记录数
originalDF.select("date").na.drop().count()

完整可运行示例代码 #

这里给你一个完整的Scala示例，涵盖上述所有注意点：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

object StreamingParquetPartitionFix {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("StreamingParquetPartition")
      .master("local[*]") // 生产环境请移除该配置
      .getOrCreate()

    import spark.implicits._

    // 替换为你的实际数据源（比如Kafka、Socket等）
    val inputStream = spark.readStream
      .format("socket")
      .option("host", "localhost")
      .option("port", 9999)
      .load()
      .selectExpr(
        "current_date() as date",
        "cast(rand()*10 as int) as zoneId",
        "concat('device_', cast(rand()*100 as int)) as deviceId",
        "cast(rand()*1000 as int) as timeSinceLast"
      )

    // 确保date是标准Date类型
    val preparedDF = inputStream.withColumn("date", to_date(col("date")))

    // 写入流：按date分区，自动剥离分区列
    val query = preparedDF.writeStream
      .format("parquet")
      .partitionBy("date") // 指定分区列
      .option("path", "/tmp/streaming_parquet_output") // 输出路径
      .option("checkpointLocation", "/tmp/streaming_parquet_checkpoint") // 新的checkpoint目录
      .outputMode("append") // 适合增量数据的输出模式
      .start()

    query.awaitTermination()
  }
}

验证分区是否生效 #

启动作业后，查看输出目录，应该能看到类似date=2024-05-20的子目录；再读取生成的Parquet文件，查看schema会发现date字段不在文件的schema中（因为它是分区列，只存在于目录路径）。

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