Spark SQL JOIN中df.repartition与读分区Parquet执行耗时差异原因问询

核心差异原因 #

• Stage与Shuffle开销不同
  DataFrame.repartition("id")属于宽依赖操作，本身就会触发一次全量Shuffle：所有数据需要经过序列化、跨节点网络传输、反序列化、落盘临时Shuffle文件的流程，这部分开销会额外生成多个Stage。加上后续Join操作对应的Stage，最终才会出现3个Job、7个Stage的执行计划，其中近2/3的耗时都消耗在了重分区的Shuffle流程上。
  而读取write.partitionBy("id")生成的Parquet文件时，Spark可以直接利用Parquet的分区元数据实现分区对齐，不需要额外执行Shuffle操作，整个流程只需要完成「读取分区数据→执行Broadcast Hash Join」两个步骤，因此只有1个Job、1个Stage，省掉了多次Shuffle的巨额开销。
• IO与数据本地性开销不同
  repartition生成的内存分区需要跨节点传输数据，网络IO开销远高于磁盘本地IO。而写入分区Parquet后，相同id的所有数据都存储在同一个分区目录下，读取时Spark可以优先调度本地节点的Executor读取本地磁盘的分区数据，不需要跨节点拉取；同时Parquet作为列式存储，支持列裁剪、谓词下推优化，可以只读取Join需要的id字段和关联字段，进一步降低了实际读取的数据量。
• Join执行效率不同
  虽然两种场景下通过df.rdd.getNumPartitions()拿到的分区数相同、均按id分区排序，但repartition生成的临时分区没有持久化的统计元数据，执行Broadcast Hash Join时需要额外计算每个分区的哈希分布、校验分区对齐状态。而分区Parquet的元数据中已经记录了每个分区的id值范围、数据量统计信息，Join时可以直接匹配两个表的对应分区，不需要额外的计算开销，同时200万行规模的小表广播可以和4100万行规模的大表并行读取同时进行，减少了串行等待时间。

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