Spark Scala写入超50GB分区CSV的性能调优咨询

针对你处理50GB+大数据、按CATEGORY_ID分区写入CSV的场景，我整理了几个实用的优化方向，结合你的代码场景逐一拆解：

1. 优化DataFrame分区，匹配写入需求 #

聚合后的DataFrame分区数如果不合理，会直接导致写入时出现大量小文件或者单个文件过大，严重拖慢性能。可以根据CATEGORY_ID的基数调整分区策略：

• 如果CATEGORY_ID基数较小（几百个以内）：
  直接按CATEGORY_ID重分区，让每个分区对应一个类别，写入时每个分区会生成对应类别的文件，减少小文件和shuffle开销：

  df.repartition(col("CATEGORY_ID"))
    .write 
    .mode(SaveMode.Overwrite) 
    .partitionBy("CATEGORY_ID") 
    .format("csv") 
    .option("header", "true") 
    .option("sep", "|") 
    .option("quoteAll", true) 
    .csv("output/inventory_backup")
  

• 如果CATEGORY_ID基数较大（上万个以上）：
  结合固定分区数和CATEGORY_ID重分区，既保证并行度，又避免生成过多小文件。比如设置分区数为集群核心数的2-3倍（示例中用200，可根据资源调整）：

  df.repartition(200, col("CATEGORY_ID"))
    .write 
    .mode(SaveMode.Overwrite) 
    .partitionBy("CATEGORY_ID") 
    .format("csv") 
    .option("header", "true") 
    .option("sep", "|") 
    .option("quoteAll", true) 
    .csv("output/inventory_backup")
  

另外，groupBy操作默认依赖spark.sql.shuffle.partitions参数（默认200），如果数据量很大，适当调大这个参数（比如500或1000），避免shuffle时数据倾斜：

spark.conf.set("spark.sql.shuffle.partitions", 500)

2. 调整CSV写入的专属优化参数 #

Spark的CSV格式有几个参数能直接提升写入效率：

• 启用压缩减少IO量：
  如果业务允许，开启压缩能大幅降低磁盘IO开销。推荐用snappy（兼顾压缩比和读写速度），或者gzip（更高压缩比）：

  .option("compression", "snappy")
  

• 限制单文件记录数：
  设置maxRecordsPerFile避免单个文件过大，同时减少零散小文件。比如限制每个文件最多100万条记录：

  .option("maxRecordsPerFile", 1000000)
  

• 优化引号策略：
  quoteAll=true会给所有字段加引号，增加处理时间和文件大小。如果不需要全量引号，改成quoteMode=MINIMAL（仅对包含分隔符、换行符的字段加引号），能显著降低写入开销：

  .option("quoteMode", "MINIMAL") // 替代quoteAll=true
  

3. 调整集群资源配置，提升并行能力 #

50GB+的数据需要足够的集群资源支撑，建议调整以下参数：

• 增加executor内存：比如设置--executor-memory 16G，避免内存不足导致的GC频繁或任务失败。
• 调整executor核心数：比如--executor-cores 4，让每个executor能并行处理更多任务。
• 增加executor数量：根据集群可用资源，适当增加executor数量，提升整体并行处理能力。
• 开启自适应执行（Spark 3.x+）：让Spark根据实际数据情况自动调整分区数、shuffle策略，优化性能：

  spark.conf.set("spark.sql.adaptive.enabled", "true")
  spark.conf.set("spark.sql.adaptive.shuffle.targetPostShuffleInputSize", "64m") // 目标分区大小可根据数据调整
  

4. 预处理阶段减少数据量 #

在聚合之前先过滤无效数据，能减少后续所有步骤的处理压力：

// 过滤空值、无效类别等数据
val filteredDf = df.filter(col("CATEGORY_ID").isNotNull && col("ASSORTED_STOCK_UNIT").isNotNull)
// 再执行聚合操作
val aggregatedDf = filteredDf.groupBy("PRODUCT_ID","LOC_ID","DAY_ID") 
  .agg(functions.sum("ASSORTED_STOCK_UNIT").as("ASSORTED_STOCK_UNIT_sum"), ...)

这些优化方向可以根据你的实际集群资源、数据基数灵活组合，优先从分区调整和资源配置入手，能快速看到性能提升。

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