如何用PySpark计算每个序列号的时长并过滤超2天数据？

实现步骤 #

假设你的数据集包含serial_number（序列号）和时间字段（比如event_time），核心逻辑是按序列号分组计算时间跨度，再筛选出跨度超过2天的序列号。以下是完整实现流程：

1. 导入依赖模块 #

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

2. 标准化时间字段类型 #

如果你的时间字段不是Timestamp类型，先做类型转换：

df = df.withColumn("event_time", F.to_timestamp("event_time"))

3. 计算每个序列号的时间跨度 #

按序列号分组，提取每组的最早、最晚时间，再计算两者的天数差：

duration_df = df.groupBy("serial_number") \
    .agg(
        F.max("event_time").alias("max_time"),
        F.min("event_time").alias("min_time")
    ) \
    .withColumn("duration_days", F.datediff(F.col("max_time"), F.col("min_time")))

4. 过滤出符合条件的序列号 #

筛选出时长超过2天的记录，可按需保留时长字段：

filtered_serials = duration_df.filter(F.col("duration_days") > 2) \
    .select("serial_number", "duration_days")

5. （可选）关联回原始数据 #

如果需要获取这些序列号对应的所有原始记录，可通过关联实现：

filtered_full_data = df.join(
    filtered_serials,
    on="serial_number",
    how="inner"
)

完整示例代码（含数据加载） #

结合你提供的加载数据逻辑，完整代码如下：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# 初始化Spark会话
spark = SparkSession.builder.appName("SerialDurationFilter").getOrCreate()
# 加载数据（替换为你的数据源路径/方式）
df = spark.read.csv("your_data.csv", header=True, inferSchema=True)

# 转换时间字段（若原始数据时间格式非Timestamp）
df = df.withColumn("event_time", F.to_timestamp("event_time"))

# 计算时长并过滤
duration_df = df.groupBy("serial_number") \
    .agg(F.max("event_time").alias("max_time"), F.min("event_time").alias("min_time")) \
    .withColumn("duration_days", F.datediff(F.col("max_time"), F.col("min_time")))

filtered_serials = duration_df.filter(F.col("duration_days") > 2).select("serial_number", "duration_days")

# 查看结果
filtered_serials.show()

关键函数说明 #

• groupBy("serial_number"): 按序列号维度分组聚合
• F.max()/F.min(): 提取每组的最晚、最早时间点
• F.datediff(): 直接计算两个日期的天数差，适配时长统计需求
• filter(): 按时长条件筛选目标序列号

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