PySpark如何对存储0/1值的多列DataFrame做聚合统计生成汇总表

你可以通过行列转置+分组聚合的方式实现需求，无需手动逐一编写20列的聚合逻辑，通用实现代码如下：

提示：如果源DataFrame包含不需要统计的非0/1列，只需要修改统计列列表的赋值逻辑，过滤掉不需要的列即可。

PySpark 实现 #

from pyspark.sql import functions as F

# 替换为你自己的源DataFrame
source_df = 你的初始DataFrame
# 获取所有需要统计的0/1列
stat_columns = source_df.columns

# 构造unpivot（行转列）表达式，将所有统计列转为[列名, 取值]的长表结构
stack_expression = f"stack({len(stat_columns)}, {','.join([f"'{col}', {col}" for col in stat_columns])}) as (col_name, Values)"

# 转长表 -> 按取值+列名分组计数 -> 再转成目标宽表格式
result_df = source_df.select(F.expr(stack_expression)) \
              .groupBy("Values") \
              .pivot("col_name") \
              .count() \
              .orderBy("Values")

# 查看结果
result_df.show()

# 若需输出为|分隔的文本文件，可执行以下代码
# result_df.write.option("sep", "|").csv("你的输出目录路径", header=True)

Scala Spark 实现 #

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

// 替换为你自己的源DataFrame
val sourceDf = 你的初始DataFrame
val statColumns = sourceDf.columns.toSeq

// 构造unpivot表达式
val stackExpression = s"stack(${statColumns.size}, ${statColumns.map(col => s"'$col', $col").mkString(",")}) as (col_name, Values)"

val resultDf = sourceDf.select(expr(stackExpression))
                 .groupBy("Values")
                 .pivot("col_name")
                 .count()
                 .orderBy("Values")

resultDf.show()

输出结果的结构完全符合要求，第一列为Values（取值0、1），后续每列为原DataFrame的字段，单元格值为对应列下0/1的出现次数。

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