基于HDFS Parquet数据集的PySpark作业资源配置咨询

一、核心测算逻辑 #

所有参数推导完全基于给定的无数据倾斜、数据均匀分布在HDFS的前提，无需通过多轮试验调整，核心原则是保证单Executor的资源利用率稳定在60%~70%，预留足够的GC、序列化、临时溢出开销，避免OOM和资源浪费。
通用测算的底层参考是：PySpark处理Parquet格式数据时，单Executor的内存占用存在固定膨胀系数（Java/Python对象序列化后大小是Parquet解压后大小的1.5~1.8倍），且Spark按分区流式处理数据，不需要一次性全量加载数据集到内存。

二、具体参数测算（结合给定数据集） #

给定的两份数据集：

• 大表：1.6T原始Parquet，解压后2.8T，31列
• 小表：200G原始Parquet，解压后360G，5列
  作业逻辑为按id关联、用小表3个字段丰富大表，全程无倾斜。

1. 单Executor的基础配置 #

EMR上PySpark作业的单Executor配置有经过验证的最优基线，无需随意调整：

• CPU核数：固定设为4核/Executor。超过8核会显著提升Full GC的停顿风险，低于2核会造成节点资源浪费，4核配置的任务调度、资源利用效率最高。
• 内存：固定按每核4G配置，即16G/Executor。其中JVM堆内存占12.8G，剩余3.2G配置为spark.executor.memoryOverhead，专门给Python worker进程、堆外内存、shuffle临时溢出预留空间，可避免PySpark进程被YARN强制杀掉。

该配置下，单Executor可稳定处理的有效数据规模约为9G（扣除30%的固定开销），不会出现内存溢出。

2. 总Executor数量测算 #

通用经验公式（适配无倾斜Parquet场景）：

总Executor数量 = （大表原始大小 / 100G） * 1.2（shuffle开销冗余）
结合本场景的大表1.6T原始大小，代入计算：
总Executor数量 = (1600G /100G)*1.2 ≈ 20个
对应总资源规模：80核CPU、320G总内存。
这个配置下，任务总调度开销适中，shuffle阶段的内存预留充足，整体作业资源利用率约65%，无浪费也不会过载。

3. Shuffle分区匹配配置 #

无倾斜场景下，Spark单任务处理的最优分区大小为128M~256M，结合本场景join阶段的总shuffle数据量（约3.2T解压后大小），直接将spark.sql.shuffle.partitions设为2400即可，保证每个shuffle分区大小在130M左右，既不会因为分区过大触发OOM，也不会因为分区过细碎增加调度开销。

注意：本场景下小表解压后规模达360G，不要手动开启broadcast join（即使调大广播阈值），sort merge join的效率远高于广播join——广播join需要把全量小表复制到每个Executor内存中，20个Executor就要重复加载7.2T的小表数据，反而会造成严重的内存浪费。

三、磁盘资源配置判断 #

无需额外配置磁盘资源。
原因：

• 本场景下shuffle阶段产生的临时磁盘数据约为总解压数据量的10%，即合计320G左右，分摊到20个Executor上，单节点仅需承载16G临时shuffle文件。
• EMR默认核心节点本地挂载的SSD磁盘容量普遍在150G以上，远大于shuffle临时文件的占用需求，完全可以覆盖。
  只有当作业涉及大表与大表shuffle、存在数据倾斜导致单节点临时磁盘占用超过本地磁盘50%时，才需要额外扩容磁盘。

四、通用测算公式总结 #

对于无数据倾斜、HDFS上的Parquet格式PySpark作业，可直接套用以下公式快速配参，无需反复试错：

• 单Executor核数：4核
• 单Executor内存：16G（含3.2G的内存overhead）
• 总Executor数量：按每100G大表原始Parquet大小配1个Executor，额外加20%的冗余
• 磁盘配置：EMR默认磁盘即可，仅当shuffle临时数据占单节点磁盘50%以上时扩容
• Shuffle分区数：按每128M单分区大小设置，即 总shuffle数据量(解压后)/128M

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