如何优化PySpark以充分利用集群所有节点的全部计算核心？

核心原因分析 #

• 完全忽略Spark固有调度开销：你将1000行数据拆分到75个分区，单个分区仅含13行左右，任务调度、序列化反序列化、进程间通信的开销占比远超实际计算开销，理想计算模型完全没有考虑这部分固定成本。
• 分布式shuffle开销未计入：repartition(75)操作会触发全量shuffle，需要将数据写入临时文件、跨节点网络传输，这部分开销在小数据量场景下占比极高，你的预期耗时完全没有覆盖这部分成本。
• 配置可能未实际生效：Google DataProc默认开启动态资源分配，你手动指定的spark.executor.instances=5会被动态分配策略覆盖，实际运行时可能没有启动足够的executor实例；同时节点上的DataNode、NodeManager等系统进程也会占用CPU资源，无法保证所有15个executor核都能无抢占执行任务。
• 函数实现存在冗余开销：你本地单线程运行时，func的初始化逻辑（比如模型加载、外部连接建立、公共参数计算）仅执行1次；改用foreachPartition后，初始化逻辑如果写在函数内部，会在75个分区各执行1次，额外带来大量重复开销。

可落地优化方案 #

• 修正Spark配置：新增配置项关闭动态资源分配，保证资源配置生效：

  conf = spark.sparkContext._conf.setAll([
      ('spark.executor.cores', '15'), 
      ('spark.executor.instances', '5'),
      ('spark.dynamicAllocation.enabled', 'false'),
      ('spark.task.cpus', '1')
  ])
  

• 调整分区数量：避免过度分区，计算密集型场景下分区数设置为总核心数的1_{2倍即可，你可以尝试将分区数改为10}20个，大幅降低调度开销占比。
• 优化函数实现：将func中的一次性初始化逻辑移到分区处理的最开头，保证每个分区仅执行一次初始化，避免重复开销。
• 减少不必要的shuffle：如果业务逻辑允许，优先用mapPartitions替代foreachPartition配合重分区的逻辑，或者在limit(1000)后先执行cache()缓存数据，降低重复扫描和shuffle的开销。
• 向量化处理优化：如果func支持批量处理，可以将分区内的行转换为Pandas DataFrame后用向量化运算处理，比逐行遍历效率提升数倍。

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