Spark中getNumPartitions与partitionBy的困惑及分区数疑问

问题解析与解答 #

两个“分区”的区别（结果都正确，对应不同概念）

• show partitions和目录中的933个子文件夹：这是表的存储分区，是Spark/Hive按channelIdVehicleId字段唯一值在磁盘上划分的目录结构，核心作用是优化查询性能（比如过滤特定车辆时仅扫描对应目录），和计算并行度无直接关联。
• rdd.getNumPartitions返回的29：这是DataFrame/RDD的计算分区，是Spark用于并行处理数据的任务单元，每个计算分区对应一个执行Task。Spark加载表时会根据文件大小、默认配置（如spark.sql.files.maxPartitionBytes默认128MB）合并小文件，所以计算分区数和存储分区数并非一一对应——多个小存储分区的数据可能被合并到一个计算分区。

mapPartitions()使用的分区数

调用mapPartitions()时，使用的是29个计算分区，而非933个存储分区。因为Spark加载表时的计算分区规划是基于数据文件大小，而非存储目录数量。

实现每台车辆单独计算的方案

要确保每个车辆对应一个计算分区，可通过以下两种方式：

1. 加载表后重分区
   加载表后，按channelIdVehicleId字段重分区，同时调整shuffle分区数匹配唯一值数量：

   // 先设置shuffle分区数为933，避免合并小分区
   spark.conf.set("spark.sql.shuffle.partitions", 933)
   val df = spark.table("dac_dev.tmp.testPartition")
                 .repartition($"channelIdVehicleId")
   // 此时df的计算分区数会对应933个车辆，调用mapPartitions即可实现单车辆单独计算
   df.mapPartitions(...)
   

2. 写入表时提前规划
   在写入表前先按channelIdVehicleId重分区，再执行分区存储，这样每个存储目录会对应一个计算分区：

   dfEnrichedTripsToPredict
     .repartition($"channelIdVehicleId")
     .write
     .partitionBy("channelIdVehicleId")
     .saveAsTable("dac_dev.tmp.testPartition")
   

   后续加载该表时，每个车辆的存储目录对应一个计算分区，直接满足单车辆单独计算的需求。

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