PySpark性能调优咨询：超大规模数据集处理提速需求

一、集群资源扩容与参数调优 #

现有集群总核心数仅40（8 Executor × 5核），总Executor内存80G，面对新增24%的数据量，资源储备已不足以支撑目标耗时，优先调整资源配置：

• 扩容Executor数量：将Executor数从8提升至16-20（需结合YARN集群剩余资源），总核心数可达80-100，直接提升并行处理能力。保持每Executor核心数5、内存10G（2G/核的配比符合Cloudera最佳实践）。
• 调整内存overhead：设置spark.executor.memoryOverhead=4g（默认1G），避免因堆外内存不足引发的OOM或频繁GC，尤其适合大shuffle、缓存场景。
• 优化shuffle分区数：原300分区针对126亿数据最优，新增3亿后可测试400-500分区，确保每个shuffle分区数据量控制在400-500M，避免单分区过大拖慢任务。
• 开启动态资源分配：设置spark.dynamicAllocation.enabled=true，让集群根据任务负载自动增减Executor，峰值时段利用闲置资源。

二、数据处理逻辑深度优化 #

• 强化谓词下推：确保过滤条件（如时间、状态筛选）在数据读取阶段执行，Parquet原生支持谓词下推，可通过spark.sql.parquet.filterPushdown=true开启，减少读取的数据量。例如：

  df = spark.read.parquet("hdfs://path").filter("dt >= '2024-01-01'")
  

• 优化聚合逻辑：若存在多维度聚合，可先按分区内维度预聚合，再全局聚合，减少shuffle数据量。例如：

  # 先分区内聚合
  pre_agg_df = df.groupBy("partition_col", "agg_col").count()
  # 再全局聚合
  final_agg_df = pre_agg_df.groupBy("agg_col").sum("count")
  

• 缓存策略精细化：仅缓存被重复使用≥2次的DataFrame，且选择MEMORY_AND_DISK_SER存储级别（序列化存储节省内存，降低GC频率）：

  from pyspark.storagelevel import StorageLevel
  df.cache(StorageLevel.MEMORY_AND_DISK_SER)
  

• 排查数据倾斜：通过Spark UI的Stage页面查看Task数据量分布，若某Task处理数据量是其他的10倍以上，即为倾斜：
  • 对大键进行加盐拆分：在大键后拼接随机后缀，拆分聚合后再合并结果；
  • 过滤异常大键：若某键数据量占比过高且非业务必需，直接过滤。

三、Parquet存储层优化 #

• 压缩算法选择：设置spark.sql.parquet.compression.codec=snappy，Snappy压缩速度远快于Gzip，牺牲少量压缩比换取处理效率，适合低延迟需求。
• 合理分区存储：按聚合维度（如日期、地域）对输出Parquet进行分区，后续读取可直接过滤分区，减少扫描数据量：

  final_agg_df.write.partitionBy("dt").parquet("hdfs://output_path")
  

• 控制文件大小：确保输出Parquet文件大小在128M-256M之间，可通过repartition或coalesce调整：

  # 估算总数据量后设置分区数，确保单文件大小达标
  final_agg_df.repartition(500).write.parquet("hdfs://output_path")
  

• 合并小文件：若输入数据源存在大量小文件（<64M），读取前先合并，避免调度开销：

  small_files_df = spark.read.parquet("hdfs://small_files_path")
  merged_df = small_files_df.repartition(100)  # 合并为100个文件
  

四、序列化与GC优化 #

• 切换Kryo序列化：替换默认JavaSerializer为Kryo，提升序列化速度与空间利用率：

  spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  spark.conf.set("spark.kryo.registrationRequired", "true")
  # 注册需序列化的类（如自定义数据类型）
  spark.sparkContext._jvm.org.apache.spark.serializer.KryoSerializer.registerClasses([...])
  

• 优化GC配置：为Executor启用G1GC垃圾回收器，减少停顿时间：

  spark.conf.set("spark.executor.extraJavaOptions", "-XX:+UseG1GC -XX:MaxGCPauseMillis=200")
  

• 替换普通UDF为Pandas UDF：Python与JVM的单条数据交互开销极大，批量处理的Pandas UDF可将效率提升5-10倍：

  from pyspark.sql.functions import pandas_udf
  from pyspark.sql.types import LongType
  
  @pandas_udf(LongType())
  def sum_udf(col):
      return col.sum()
  
  df.select(sum_udf("value"))
  

五、任务调度优化 #

• 开启任务推测执行：设置spark.speculation=true，自动检测并重启慢任务，避免单个节点故障或负载过高拖慢整个Stage。
• 调整Task并行度：确保spark.default.parallelism设置为总核心数的2-3倍（如16 Executor×5核=80核心，设置为160-240），提升Task并行处理能力。

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