如何在Spark中保留Parquet原分区并高效完成转换写入？

要实现按原分区方式（sensor_name）高效存储转换后的数据，核心是避免Spark在写入时触发不必要的Shuffle，关键在于确保读取后的每个RDD物理分区内仅包含单一sensor_name的数据，这样Spark在执行partitionBy写入时会直接将分区数据写入对应文件夹，无需跨分区洗牌。

具体步骤 #

1. 正确读取原分区Parquet数据
   Spark读取分区Parquet表时会自动识别sensor_name为分区列，且默认会将每个分区文件夹内的文件映射为独立的RDD物理分区（每个分区内的sensor_name值唯一）。直接读取即可保留原分区结构：

# Python示例
df = spark.read.parquet("原Parquet文件夹路径")

// Scala示例
val df = spark.read.parquet("原Parquet文件夹路径")

2. 执行窄依赖转换操作
   你的转换（如reading * 10）属于窄依赖操作（仅对单条数据做列计算，无需跨分区交互），不会改变原物理分区结构，每个分区内的sensor_name仍保持唯一：

transformed_df = df.withColumn("reading", df["reading"] * 10)

val transformedDf = df.withColumn("reading", col("reading") * 10)

3. 直接按分区列写入
   此时直接调用partitionBy("sensor_name")写入，Spark会检测到每个物理分区内的sensor_name值唯一，无需触发Shuffle，直接将分区数据写入对应sensor_name文件夹：

transformed_df.write
    .format("parquet")
    .partitionBy("sensor_name")
    .mode("overwrite")
    .save("输出路径")

transformedDf.write
    .format("parquet")
    .partitionBy("sensor_name")
    .mode("overwrite")
    .save("输出路径")

为什么之前的方法速度慢？ #

• 仅用partitionBy未确保分区单一性：如果读取后的RDD分区包含多个sensor_name的数据，Spark会触发Shuffle将相同sensor_name的数据聚合到一起，这是开销最大的环节。
• 提前repartition("sensor_name")：会主动触发Shuffle重新分区，虽然后续写入时无需再Shuffle，但额外的Shuffle操作仍会增加耗时——原数据已经是按sensor_name分区的结构，这一步完全没必要。

额外优化建议 #

• 避免合并小文件：如果原分区文件夹内有大量小文件，可通过spark.sql.files.maxPartitionBytes调整分区大小，确保每个RDD分区对应合理大小的文件（默认128MB），但不要设置过大导致跨sensor_name合并分区。
• 关闭分区列类型推断：添加配置spark.sql.sources.partitionColumnTypeInference.enabled=false，避免Spark自动推断分区列类型带来的额外开销（可选）。

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