Spark如何优化超大量列DataFrame按行值统计各列计数的效率

核心问题定位 #

你现有方案的性能瓶颈来自逐列触发count action，7万列相当于提交7万次Spark作业，分布式调度的 overhead 被无限放大，自然耗时极长。 #

可落地的优化手段 #

• 单次作业全列聚合（最推荐，适配绝大多数场景）
  利用Spark的agg算子一次性完成所有列的统计，全程只触发1次作业，耗时可以从小时级降到秒级。因为你要统计的是值为1的数量，直接对列求和即可得到对应计数：

  import org.apache.spark.sql.functions.{col, sum}
  // 生成所有列的统计表达式
  val countExpr = df.columns.map(c => sum(col(c)).alias(s"${c}_1_cnt"))
  // 单次聚合得到所有列的统计结果
  val result = df.agg(countExpr.head, countExpr.tail: _*)
  

• 小数据集转本地计算
  你的数据集仅10k行，全量拉到Driver端本地计算的开销远低于分布式调度开销，不需要走分布式任务流程：

  val localData = df.collect()
  val col1Count = df.columns.map(c => localData.count(row => row.getAs[Int](c) == 1))
  

• 转长表后聚合（适合高频统计场景）
  如果这类列统计是你的高频需求，可以先把宽表转成<列名, 列值>的长表存储，后续统计直接按列名分组即可，不需要每次生成大量聚合表达式：

  // 宽表转长表
  val longDf = df.selectExpr(
    s"stack(${df.columns.size}, ${df.columns.map(c => s"'$c', $c").mkString(",")}) as (col_name, col_value)"
  )
  // 统计值为1的列计数
  val result = longDf.filter(col("col_value") === 1).groupBy("col_name").count()
  

• 参数适配优化
  如果你使用分布式计算方案，可将spark.sql.shuffle.partitions调整为10~20（默认是200），减少小数据集场景下不必要的shuffle分区开销。

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