Spark DataFrame如何统计各列on值数量并生成指定格式结果表

实现方法 #

两种高效的写法，都可以直接生成符合要求的结果DataFrame，不需要逐列手动统计：

方法1：用stack函数转置统计（推荐，全分布式执行，性能最优）

核心思路是先把宽表转成「列名-列值」的长表，再过滤统计，全程不触发Driver端的collect操作，适合大数据量、多列的场景：

import org.apache.spark.sql.functions._

// 可手动指定要统计的列，例如val targetCols = Array("A", "B", "C")
val targetCols = df_original.columns
// 构造转置用的stack表达式
val stackExpr = s"stack(${targetCols.size}, ${targetCols.map(colName => s"'$colName', $colName").mkString(", ")}) as (Features, value)"

val resultDF = df_original.selectExpr(stackExpr)
  .filter(col("value") === "on")
  .groupBy("Features")
  .count()
  .withColumnRenamed("count", "Count_on")

方法2：逐列聚合后转置（适合列数较少的场景）

先对每列单独统计on的数量，得到单行的聚合结果后再转成目标格式：

import org.apache.spark.sql.functions._

val targetCols = df_original.columns
// 生成每列的on计数聚合表达式
val aggExprs = targetCols.map(colName => sum(when(col(colName) === "on", 1).otherwise(0)).alias(colName))
// 执行聚合得到单行结果
val aggResult = df_original.agg(aggExprs.head, aggExprs.tail: _*).head()
// 转置为目标DataFrame
val resultDF = spark.createDataFrame(targetCols.map(colName => (colName, aggResult.getAs[Long](colName))))
  .toDF("Features", "Count_on")

你原来的逐列collect的写法会多次触发Spark作业，列数多的时候效率极低，且手动编码成本高，以上两种方法都可以自动适配所有待统计列，不需要修改代码就能兼容列的增减。

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