PySpark中数据按小时聚合及evtTime小时提取需求

没问题，我来帮你搞定PySpark里按小时聚合数据的需求！咱们一步步来，先处理evtTime时间字段，再完成聚合操作：

1. 转换字符串时间为Timestamp类型 #

你的evtTime字段是字符串格式（比如2017-10-01 23:03:51.337），首先得把它转成PySpark的Timestamp类型，这样后续的时间操作才更可靠。用to_timestamp函数指定匹配的时间格式即可：

from pyspark.sql.functions import to_timestamp, hour, date_format, sum

# 新增Timestamp类型的时间列
df = df.withColumn("evt_timestamp", to_timestamp(df.evtTime, "yyyy-MM-dd HH:mm:ss.SSS"))

2. 提取小时维度 #

这里有两种常用的提取方式，根据你的业务需求选：

• 仅提取小时数（0-23）：如果不需要区分日期，只看一天中的小时段，用hour()函数：

  df = df.withColumn("evt_hour", hour(df.evt_timestamp))
  

• 提取日期+小时（推荐）：如果要避免不同日期的同一小时被合并（比如10月1日23点和10月2日23点要分开统计），用date_format转成yyyy-MM-dd HH格式：

  df = df.withColumn("evt_date_hour", date_format(df.evt_timestamp, "yyyy-MM-dd HH"))
  

3. 按小时维度执行聚合 #

假设你需要统计每个小时段的event_count总和，用groupBy配合聚合函数就能实现：

• 按日期+小时聚合：

  hourly_agg_df = df.groupBy("evt_date_hour").agg(sum("event_count").alias("total_events"))
  

• 按单独小时数聚合：

  hourly_agg_df = df.groupBy("evt_hour").agg(sum("event_count").alias("total_events"))
  

你可以根据实际需求替换聚合函数，比如用count()统计事件数量、avg()计算平均耗时等。如果你的evtTime字段本身已经是Timestamp类型，那第一步可以直接跳过~

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