如何在PySpark DataFrame中生成全时段30分钟间隔的时间戳列

1. 处理原数据的时间戳格式 #

如果你的interval_date_time列是字符串类型，先转换成Timestamp类型，方便后续日期操作：

from pyspark.sql import functions as F
from pyspark.sql.types import TimestampType

# 假设原DataFrame名为df
df = df.withColumn("interval_date_time", F.to_timestamp("interval_date_time", "yyyy-MM-dd:HH:mm:ss"))

2. 提取原数据涉及的所有唯一日期 #

从时间戳列中提取出所有不重复的日期（仅保留年月日部分）：

unique_dates = df.select(F.date_trunc("day", "interval_date_time").alias("date")).distinct()

3. 为每个日期生成完整的30分钟间隔 #

利用Spark的sequence函数，为每个日期生成从当天00:00:00到23:30:00的每30分钟时间戳，再通过explode将数组展开为单行记录：

full_intervals = unique_dates.withColumn(
    "start_time", F.date_trunc("day", "date")  # 当天00:00:00
).withColumn(
    "end_time", F.date_add(F.date_trunc("day", "date"), 1) - F.expr("INTERVAL 30 MINUTES")  # 当天23:30:00
).withColumn(
    "interval_date_time", F.explode(F.sequence("start_time", "end_time", F.expr("INTERVAL 30 MINUTES")))
).select("interval_date_time")

4. 可选：关联原数据补全其他字段 #

如果需要保留原数据的其他字段，用左连接将生成的完整时间戳与原数据关联，缺失字段会自动填充为null：

result_df = full_intervals.join(df, on="interval_date_time", how="left")

注意事项 #

• 确保Spark版本在2.4及以上，sequence函数从该版本开始支持
• 若原时间戳列已是Timestamp类型，可跳过第一步的格式转换

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