PySpark/Pandas按人员分组计算各活动的持续时长

以下分别给出Pandas和PySpark两种框架的实现代码，实现前请先确认Timeframe字段存储的时间格式可被正常解析，两种实现均默认按人员独立分组、同组内按时间先后顺序匹配下一条记录的时间点做差，无后续记录的活动时长标记为N/A。

Pandas 实现 #

核心逻辑是通过groupby+shift(-1)直接取同组下一行的时间值，和当前行时间做差即可：

import pandas as pd

# 转换时间字段为标准datetime类型
df["Timeframe"] = pd.to_datetime(df["Timeframe"])

# 按人员分组，匹配每条记录对应的下一个时间点
df["next_time"] = df.groupby("Person")["Timeframe"].shift(-1)

# 计算时间差，无下一时间点的记录填充N/A
# 若需要输出分钟/小时等数值单位，可对Timedelta做单位转换，例如.dt.total_seconds()//60 得到分钟数
df["Duration"] = (df["next_time"] - df["Timeframe"]).fillna("N/A")

# 删除临时辅助列
df = df.drop(columns=["next_time"])

PySpark 实现 #

核心逻辑是通过窗口函数lead()取同分区内排序后的下一行时间值做差值计算：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 转换时间字段为标准timestamp类型
df = df.withColumn("Timeframe", F.to_timestamp(F.col("Timeframe")))

# 定义窗口规则：按人员分区，同分区内按时间升序排列
person_window = Window.partitionBy("Person").orderBy("Timeframe")

# 匹配下一时间点、计算时长，空值填充N/A
df = df.withColumn(
    "next_time",
    F.lead(F.col("Timeframe"), 1).over(person_window)
).withColumn(
    "Duration",
    F.when(
        F.col("next_time").isNotNull(),
        # 示例以分钟为单位，转小时则除以3600即可
        (F.col("next_time").cast("long") - F.col("Timeframe").cast("long")) / 60
    ).otherwise(F.lit("N/A"))
).drop("next_time")

说明：由于题目明确所有活动之间无时间间隔，因此相邻时间点的差值即为当前活动的准确持续时长，无需额外处理间隔数据。如果需要将时长格式化为X小时Y分钟的可读形式，可基于计算得到的时间差数值自行做格式化转换。

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