PySpark处理34GB大固定宽文件性能优化求助

一、数据读取阶段优化 #

• 调整分区策略：默认分区逻辑可能不匹配34GB文件的体量，手动设置spark.sql.files.maxPartitionBytes（建议设为256MB或512MB），或读取后用repartition()/coalesce()调整分区数，确保每个分区大小控制在128-256MB区间，避免小分区过多或大分区拖慢并行效率。
• 使用原生固定宽读取器：不要用自定义substring逐列切割，Spark内置了固定宽格式的优化实现，示例代码如下：

  from pyspark.sql.types import StructType, StructField, StringType
  
  # 提前定义好5000列的Schema，示例仅展示结构
  custom_schema = StructType([
      StructField("col1", StringType(), nullable=True),
      StructField("col2", StringType(), nullable=True),
      # ... 其余4998列的定义
  ])
  
  df = spark.read.format("csv")
    .option("sep", "\0")
    .option("header", False)
    .option("fixedWidth", True)
    .option("widths", "10,20,...")  # 按顺序传入每列的宽度，逗号分隔
    .schema(custom_schema)
    .load("path/to/your/file")
  

  原生实现基于Java优化，比Python层面的substring循环高效数倍。

二、执行阶段优化 #

• 禁用Schema推断：提前明确指定完整的StructType Schema，避免Spark多次扫描文件推断数据类型，减少IO和计算开销。
• 列裁剪与谓词下推：如果后续业务不需要全部5000列，读取时直接筛选所需列；若有过滤条件，尽量在读取阶段就应用，减少需要处理的数据量。
• 优化Executor资源配置：在Kubernetes Operator配置中调整：
  • 每个Executor分配8-16GB内存，避免频繁GC；
  • 设置spark.executor.cores为2-4核，避免单Executor核数过多导致资源竞争；
  • 增加Executor数量，充分利用集群空闲资源，提升并行处理能力。
• 开启自适应执行：确保spark.sql.adaptive.enabled=true（Spark 3.x默认开启），让Spark根据运行时情况自动调整分区数、执行计划；同时开启spark.sql.adaptive.coalescePartitions.enabled=true，自动合并小分区。

三、Parquet存储阶段优化 #

• 配置压缩与文件大小：
  • 设置spark.sql.parquet.compression.codec=snappy，兼顾压缩比与读写速度；
  • 写入前用repartition()调整分区数，确保生成的Parquet文件大小在128-256MB之间，避免过多小文件。
• 启用向量化读写：确认spark.sql.parquet.enableVectorizedReader=true（默认开启），通过向量化操作提升Parquet的读写效率。

四、其他优化点 #

• 优化存储IO：确保文件存储在与集群同区域的存储服务，避免跨区域IO延迟；若使用本地存储，优先选择SSD磁盘。
• 版本与bug检查：确认Spark 3.1.1是否存在固定宽读取相关的已知bug，必要时升级到3.1.x的补丁版本或更高稳定版本（如3.2.x），新版本通常包含性能优化。
• 小数据量预测试：先用1GB左右的样本文件验证优化方案，确认性能提升后再全量运行，避免全量执行浪费时间。

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