Hive分区、Spark分区与Spark Join的关联关系探究

问题1：Spark读取Hive分区表后，数据集会有多少个分区？ #

咱们先把这个问题说透：当你用spark.table("table1").as[Table1Row]这种方式读取Hive分区表时，Spark生成的分区数和Hive的分区数不是直接划等号的，得看这几个关键因素：

• Hive分区下的文件细节：每个Hive分区（就是S3上date=<yyyy-MM-dd>那个目录）里有多少个文件、单个文件多大，这是基础影响因素。
• Spark的文件分片规则：Spark有个默认参数spark.sql.files.maxPartitionBytes（默认值128MB）。简单说，Spark会把单个文件或者多个小文件打包成一个Spark分区，只要总大小不超过这个阈值；要是单个文件比128MB大，就会被拆成多个Spark分区。
• 有没有做分区裁剪：要是你后续加了过滤条件（比如where date='2024-01-01'），Spark会只拉取对应Hive分区的文件，这时候Spark分区数就只和这个Hive分区下的文件分片有关了。

给你举几个真实场景的例子：

• 某个Hive分区下有3个100MB的文件：每个文件单独成一个Spark分区，这个Hive分区对应3个Spark分区；
• 某个Hive分区下有10个10MB的小文件：Spark会把它们合并成一个Spark分区（总大小100MB，没超128MB的阈值）；
• 某个Hive分区下有个200MB的大文件：Spark会把它拆成2个Spark分区（128MB+72MB）。

另外提个小坑：如果你的Hive外部表，S3上的分区目录和Hive元数据不一致（比如S3新增了分区但Hive没同步），那Spark可能读不到新分区，这时候得先跑MSCK REPAIR TABLE table1同步元数据才行。

重点：Join场景下的分区关联问题 #

既然你最终关注的是Join，那咱们聊聊这部分的核心点：

1. 分区裁剪是Join性能的关键 #

因为两个表都按date分区，只要你在Join时加上date的过滤条件，Spark会先给两个表做分区裁剪——只读取你需要的那个（或几个）Hive分区的数据，直接砍掉大部分不需要处理的数据，性能提升特别明显。比如：

val joinedData = table1.filter($"date" === "2024-01-01")
                       .join(table2.filter($"date" === "2024-01-01"), "id")

这种情况下，Spark只会处理2024-01-01这个分区的数据，Join的计算量直接缩水N倍。

2. Spark分区对齐影响Join效率 #

如果裁剪后，两个表的Spark分区数不一致，或者分区内的Join键分布不均，很容易出问题：

• 最优情况：两个表对应Hive分区的Spark分区数一致，而且每个分区里的Join键分布均匀，这时候Spark可以根据表的大小选择最适合的Join策略——小表用Broadcast Join（直接把小表广播到每个节点），大表用Sort Merge Join（排序后合并，shuffle开销小），性能拉满。
• 踩坑场景：要是某个Hive分区下的某个Join键数据量特别大（比如某个id对应100万条数据），会导致对应的Spark分区在Join时成为瓶颈，整个任务卡在这里。这时候就得做倾斜处理，比如给倾斜的键加盐拆分、过滤异常值之类的操作。

3. 写入动态分区表的Join结果要注意 #

如果Join后的数据要写入Hive动态分区表，得注意两个点：一是要设置spark.sql.sources.partitionOverwriteMode=dynamic，避免不小心覆盖整个分区目录；二是要保证Join后的数据集分区键和Hive表的分区键完全一致，不然会出现分区数据乱套的情况。

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