PySpark GroupBy性能异常排查求助（含SparkUI及调度延迟信息）

一、从SparkUI核心指标入手定位瓶颈 #

• Stage与任务分布对比：对比带WHERE条件和全量查询的Stage数量、单Stage任务数。全量查询若Stage数暴增，大概率是Shuffle分区数不合理或数据量陡增导致任务拆分过载；若单Stage任务数过少，会造成每个Task处理数据量过大，触发GC或拉长执行时间。
• Shuffle指标深度分析：重点看Shuffle Read/Write总数据量，以及单个Task的Shuffle数据分布。如果总Shuffle量是带WHERE查询的数倍，Shuffle的IO和网络传输就是核心瓶颈；若存在少数Task的Shuffle数据远超均值，直接指向数据倾斜。
• Task执行时间排查：查看Task执行时间的分布，是否有大量长尾Task（执行时间是均值的5倍以上），或是整体Task执行时间普遍拉长。前者优先排查数据倾斜，后者则可能是资源不足或GC问题。
• 调度延迟拆解：在Jobs页面查看调度延迟占总执行时间的比例，以及Stage的等待时长。若调度延迟过高，先确认集群是否有其他任务抢占资源，或是Worker节点的CPU/内存已饱和，导致Task无法及时分配启动。

二、分区与数据读取逻辑验证 #

• 确认分区扫描合理性：用explain extended查看全量查询的物理执行计划，确认是否正确遍历了所有month/day分区。若出现误扫不存在的分区，会额外增加IO负载。
• 检查Parquet文件大小分布：用hdfs dfs -du -h查看各分区下的文件大小，若存在单个文件超过1GB的超大文件，会导致对应Task执行时间过长，拖慢整体进度。
• 调整数据读取并行度：对比带WHERE查询的Task数，全量查询的Task数若过少，说明并行度不足，可调整spark.default.parallelism；若过多，会增加调度开销，需结合集群资源合理设置。

三、GC与Worker资源瓶颈排查 #

• 分析Executor GC日志：在Executors页面点击对应Executor的Logs链接，查看GC时间占总执行时间的比例。若占比超过20%，说明内存不足或GC策略不合理，可尝试增大spark.executor.memory，或改用G1GC（通过spark.executor.extraJavaOptions="-XX:+UseG1GC -XX:MaxGCPauseMillis=200"配置）。
• 监控Worker资源使用率：查看Executors页面的CPU、内存使用率。全量查询时若内存使用率接近100%，说明内存过载引发频繁GC；若CPU使用率低于50%，则可能是IO瓶颈（数据读取或Shuffle）或并行度不足导致资源浪费。
• 对比两种查询的内存负载：全量查询每个Executor处理的数据量是带WHERE查询的2倍，需确认是否超过Executor内存承载上限，导致内存溢出或GC频繁触发。

四、数据倾斜专项排查 #

• 检查GroupBy Key的分布：执行SELECT your_groupby_key, COUNT(*) FROM your_table GROUP BY your_groupby_key ORDER BY COUNT(*) DESC LIMIT 10，看是否存在热点Key（某Key的记录数是其他Key的10倍以上）。
• 验证Shuffle倾斜：在Stage页面点击Shuffle Read的详情链接，查看单个Task的Shuffle读取量，若少数Task的读取量远超其他Task，即可确认是数据倾斜导致的长尾任务。

五、针对性优化建议 #

• 数据倾斜优化：采用加盐法给热点Key添加随机后缀，拆分聚合任务；或先按分区内预聚合，再做全局聚合；开启spark.sql.adaptive.enabled让Spark自动识别并处理倾斜。
• Shuffle瓶颈优化：增大spark.sql.shuffle.partitions（默认200，可调整为1000-2000）；开启spark.shuffle.compress压缩Shuffle数据；若集群支持，使用本地磁盘作为Shuffle存储降低网络开销。
• GC问题优化：调高Executor堆内存，调整GC参数减少停顿；移除不必要的cache操作，降低内存占用。
• 调度延迟优化：开启spark.dynamicAllocation.enabled让Spark动态申请Executor；协调集群资源，避免其他大任务与当前查询抢占资源。

内容的提问来源于stack exchange，提问作者David Sánchez