PySpark按时间戳与Source分组计算每5分钟错误率KPI的实现方法

你可以通过groupBy+pivot组合实现需求，该方案原生支持数千个Source的行转列场景，适配百万级数据量处理：

实现代码 #

from pyspark.sql import functions as F

# 1. Error字段数值转换（已实现可跳过）
df = df.withColumn("error_num", F.when(F.col("Error") == "Yes", 1).otherwise(0))

# 2. 按时间窗口分组，透视Source维度计算错误率
result_df = df.groupBy("timestamp_rounded") \
              # 透视Source字段，将不同Source值转为独立列
              .pivot("Source") \
              # 聚合计算错误率
              .agg(F.avg("error_num")) \
              # 可选：无数据的时间窗口+Source组合默认填充0
              .na.fill(0)

# 3. 统一列名格式，添加Error_rate_前缀
for col_name in result_df.columns:
    if col_name != "timestamp_rounded":
        result_df = result_df.withColumnRenamed(col_name, f"Error_rate_{col_name}")

# 查看结果
result_df.show()

性能优化建议 #

当Source数量达到数千级时，建议提前给pivot函数指定Source的取值列表，避免Spark额外扫描全表计算去重的Source值，能大幅提升执行效率：

# 提前获取所有Source的取值列表（Source数量不多时可直接拉取到驱动端）
source_list = [row["Source"] for row in df.select("Source").distinct().collect()]

# 给pivot传入第二个参数指定取值范围
result_df = df.groupBy("timestamp_rounded") \
              .pivot("Source", source_list) \
              .agg(F.avg("error_num")) \
              .na.fill(0)

内容的提问来源于stack exchange，提问作者Benoît Carlier