如何高效对PySpark DataFrame按ID分组并按日重采样聚合？

针对你需要处理的Spark DataFrame重采样聚合需求——将含ID、TIMESTAMP、CONSUMPTION和TEMPERATURE列的数据集从15分钟间隔转为日间隔，按每个唯一ID对消费和温度值求和，且要适配超大数据集、避免toPandas()和循环操作，以下是最优实现方案：

核心实现代码 #

from pyspark.sql.functions import date_trunc, sum, col

# 1. 将原始时间戳截断到日级别，实现时间对齐
# 2. 按ID和日级时间戳分组，对目标列执行求和聚合
result_df = input_spark_df.withColumn("TIMESTAMP", date_trunc("day", col("TIMESTAMP"))) \
    .groupBy("ID", "TIMESTAMP") \
    .agg(
        sum("CONSUMPTION").alias("CONSUMPTION"),
        sum("TEMPERATURE").alias("TEMPERATURE")
    ) \
    .orderBy("ID", "TIMESTAMP")

代码细节说明 #

• date_trunc("day", col("TIMESTAMP"))：把15分钟粒度的时间戳统一截断到当日0点（比如2022-12-28 14:15:00会转为2022-12-28 00:00:00），完成日级重采样的时间维度对齐。
• groupBy("ID", "TIMESTAMP")：按用户ID和日级时间戳分组，确保每个ID的每日聚合结果独立计算。
• agg(...)：对分组内的CONSUMPTION和TEMPERATURE执行求和操作，并用alias保留原列名，和目标输出结构匹配。
• orderBy(...)：可选操作，让结果按ID和时间排序，与你提供的desired_outcome_spark_df顺序一致。

结果验证 #

你可以通过以下代码确认输出与目标结果一致：

# 查看结果数据
result_df.show()

# 与desired_outcome_spark_df做一致性校验
from pyspark.sql.functions import date_trunc
desired_aligned = desired_outcome_spark_df.withColumn("TIMESTAMP", date_trunc("day", col("TIMESTAMP")))
# 若返回0则说明结果完全匹配
print(result_df.exceptAll(desired_aligned).count())

效率优势说明 #

该方案完全基于Spark分布式计算引擎优化：

• 所有操作均为矢量化批量处理，无Python循环的性能损耗；
• 避免了toPandas()将分布式数据拉取到单机的内存瓶颈，可轻松支撑TB级超大数据集；
• Spark会自动对分组聚合操作做Shuffle优化、分区调整等底层优化，进一步提升处理速度。

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