Spark按IP、Timestamp聚合后将统计值转为列的简便实现方法

直接用Spark原生的pivot行转列算子就能实现，全程不需要逐列手动加列，逻辑仅需3步：

• 第一步先聚合到IP+时间戳的细粒度，计算每个分组下两个字段的去重计数
• 第二步以IP为分组维度、Timestamp为枢轴列做行转列，算子会自动为所有出现过的时间戳生成对应统计列，不需要手动枚举
• 最后统一填充空值为0即可

代码示例（PySpark实现，Scala逻辑完全一致） #

from pyspark.sql import functions as F

# 1. 先聚合得到每个IP、每个时间戳下的两个去重统计值
base_agg_df = raw_df.groupBy("IP", "Timestamp")\
    .agg(
        F.countDistinct("user_agent").alias("user_agent_cnt"),
        F.countDistinct("hostname").alias("hostname_cnt")
    )

# 2. pivot行转列，自动生成所有时间戳对应的统计列
# 注：如果提前知道所有Timestamp的取值列表，可以作为第二个参数传入pivot，大幅提升运行性能，例：.pivot("Timestamp", ts_list)
wide_df = base_agg_df.groupBy("IP")\
    .pivot("Timestamp")\
    .agg(
        F.first("user_agent_cnt").alias("user_agent_distinct"),
        F.first("hostname_cnt").alias("hostname_distinct")
    )\
    .fillna(0) # 无对应记录的位置统一填充为0

生成的宽表会自动按{时间戳值}_user_agent_distinct、{时间戳值}_hostname_distinct的规则命名列，完全符合需求。如果需要调整列顺序，比如把同一时间戳的两个指标放在相邻位置，只需要对列名排序后统一select即可，不需要新增列：

# 调整列顺序
sorted_columns = ["IP"] + sorted([col for col in wide_df.columns if col != "IP"])
final_df = wide_df.select(sorted_columns)

方案优势 #

• 没有逐列调用withColumn的冗余代码，不管时间戳有多少个取值，代码逻辑不需要修改
• 空值填充一步完成，不需要逐列处理
• 全部用Spark原生算子实现，没有额外的UDF或者循环开销，性能最优

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