PySpark全列unpivot后批量分组统计值计数实现方案

核心方案是使用Spark内置的stack函数一次性完成所有字段的逆透视转换，再通过单次分组聚合完成全字段取值计数，避免逐列遍历带来的多次全表扫描开销，执行效率远高于逐列groupBy的实现方式。

实现步骤 #

• 提取所有需要统计的目标字段（排除主键rm_ky）
• 构造stack表达式，将宽表的多列结构打平为字段名-字段值的两行长表结构，完成逆透视
• 按字段名、字段值两个维度分组，聚合统计出现次数，直接输出统一格式的结果

代码示例 #

PySpark 实现 #

from pyspark.sql import functions as F

# 读取管道符分隔的源数据，替换为你的数据路径即可
# df = spark.read.option("delimiter", "|").option("header", "true").csv("your_data_path")

# 过滤得到所有需要统计的flag字段
flag_columns = [col for col in df.columns if col != "rm_ky"]

# 构造stack逆透视表达式
stack_clause = f"""
stack(
    {len(flag_columns)},
    {','.join([f"'{col}', {col}" for col in flag_columns])}
) as (flag_name, flag_value)
"""

# 执行转换与聚合
result = df.select(F.expr(stack_clause)) \
    .groupBy("flag_name", "flag_value") \
    .count() \
    .withColumnRenamed("count", "counts")

# 输出结果
result.show()

Scala Spark 实现 #

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

// 读取管道符分隔的源数据
// val df = spark.read.option("delimiter", "|").option("header", "true").csv("your_data_path")

// 过滤得到所有需要统计的flag字段
val flagColumns = df.columns.filter(_ != "rm_ky")

// 构造stack逆透视表达式
val stackClause = s"""
stack(
    ${flagColumns.length},
    ${flagColumns.map(c => s"'$c', $c").mkString(",")}
) as (flag_name, flag_value)
"""

// 执行转换与聚合
val result = df.select(expr(stackClause))
  .groupBy("flag_name", "flag_value")
  .count()
  .withColumnRenamed("count", "counts")

// 输出结果
result.show()

注意事项 #

• 该方案仅会对源数据执行一次全表扫描，100个字段量级下性能相比逐列循环有数量级提升
• 如果待统计字段数据类型不统一，可提前将所有字段转换为兼容类型（如统一转为string）后再传入stack表达式，避免类型不匹配报错
• 如果需要按rm_ky维度做分组统计，只需要在select阶段保留rm_ky字段，分组时将rm_ky加入分组键即可

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