如何用PySpark将数据表列转为行生成统计汇总表？

方法一：逐列统计后合并 #

适合数据量较小的场景，逻辑直观易懂：

from pyspark.sql.functions import col, countDistinct, sum, when, lit

# 先获取DataFrame总行数
total_rows = df.count()

# 遍历每一列生成对应统计行
stats_list = []
for col_name in df.columns:
    col_stats = df.select(
        lit(col_name).alias("attribute_name"),
        lit(total_rows).alias("count"),
        countDistinct(col(col_name)).alias("distinct_count"),
        sum(when(col(col_name).isNull(), 1).otherwise(0)).alias("null_count")
    ).first()
    stats_list.append(col_stats)

# 转换为最终统计DataFrame
result_df = spark.createDataFrame(stats_list)

# 查看结果
result_df.show()

方法二：聚合后转置（高效版） #

适合大数据量场景，仅需一次聚合操作，性能更优：

from pyspark.sql.functions import stack, lit, countDistinct, sum, when

total_rows = df.count()

# 第一步：批量聚合所有列的去重数和空值数
aggregated_df = df.agg(
    *[countDistinct(col_name).alias(f"distinct_{col_name}") for col_name in df.columns],
    *[sum(when(col(col_name).isNull(), 1).otherwise(0)).alias(f"null_{col_name}") for col_name in df.columns]
)

# 第二步：用stack函数将宽表转置为目标窄表
num_columns = len(df.columns)
result_df = aggregated_df.select(
    stack(
        num_columns,
        *[(lit(col_name), lit(total_rows), col(f"distinct_{col_name}"), col(f"null_{col_name}")) for col_name in df.columns]
    ).alias("attribute_name", "count", "distinct_count", "null_count")
)

# 查看结果
result_df.show()

结果验证 #

针对你提供的示例数据，运行上述代码后会得到如下结果：

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