基于UNIX时间戳按subs_no和year分组计算时长遇空值问题

需求：按相同的subs_no和year分组，计算每条记录与分组内最早时间的时长差。注意new_ts列是UNIX时间戳而非字符串。

用户编写的PySpark代码如下：

from pyspark.sql import functions as F, Window as W

df_subs_loc_movmnt_ts = df_subs_loc_movmnt.withColumn("new_ts",from_unixtime(unix_timestamp(col("ts"),"HH:mm:ss"),"HH:mm:ss"))
w = W.partitionBy('subs_no', 'year').orderBy('new_ts')
df_subs_loc_movmnt_duration = df_subs_loc_movmnt_ts.withColumn('duration', F.regexp_extract('new_ts' - F.min('new_ts').over(w),"HH:mm:ss", 0))

执行后生成的df_subs_loc_movmnt_duration中duration列始终为null，数据示例：

+--------+---------------+--------+---------------+-------------+----+-----+---+--------+--------+
| date_id|             ts| subs_no|            cgi|       msisdn|year|month|day|  new_ts|duration|
+--------+---------------+--------+---------------+-------------+----+-----+---+--------+--------+
|20200801|17:00:10.000000|10100559|510-11-610104-9|3729882521647|2022|    6|  1|17:00:10|    null|
|20200801|17:09:39.000000|10100559|510-11-610104-9|3729882521647|2022|    6|  1|17:09:39|    null|
|20200801|06:44:55.000000|10100559|510-11-610034-6|3729882521647|2022|    6|  1|06:44:55|    null|
|20200801|17:45:35.000000|10100559|510-11-610079-5|3729882521647|2022|    6|  1|17:45:35|    null|
|20200801|17:48:05.000000|10100559|510-11-610660-4|3729882521647|2022|    6|  1|17:48:05|    null|
|20200801|18:07:25.000000|10100559|510-11-610127-6|3729882521647|2022|    6|  1|18:07:25|    null|
+--------+---------------+--------+---------------+-------------+----+-----+---+--------+--------+

问题原因 #

1. 字符串不能直接做减法运算：new_ts是HH:mm:ss格式的字符串，直接用new_ts - F.min('new_ts').over(w)会得到无效值，后续regexp_extract无法匹配到目标格式，因此返回null。
2. 时间处理冗余且逻辑错误：将ts转成UNIX时间戳再转回时分秒字符串完全没必要，且unix_timestamp在无日期参数时会默认使用当前日期，可能导致时间差计算偏差。

正确实现代码 #

场景1：ts是时分秒字符串（如示例中的17:00:10.000000）

先将时间字符串解析为当天的总秒数，计算差值后再转换为HH:mm:ss格式：

from pyspark.sql import functions as F, Window as W

# 将ts字符串转换为当天的总秒数（单位：秒）
df_processed = df_subs_loc_movmnt.withColumn(
    "ts_total_seconds",
    F.hour(F.col("ts")) * 3600 + F.minute(F.col("ts")) * 60 + F.second(F.col("ts"))
)

# 定义分组窗口：按subs_no和year分组
window_spec = W.partitionBy("subs_no", "year")

# 计算每条记录与分组内最早时间的秒数差，再转为HH:mm:ss格式
df_result = df_processed.withColumn(
    "duration_seconds",
    F.col("ts_total_seconds") - F.min("ts_total_seconds").over(window_spec)
).withColumn(
    "duration",
    F.concat(
        F.lpad(F.floor(F.col("duration_seconds") / 3600), 2, "0"),
        F.lit(":"),
        F.lpad(F.floor((F.col("duration_seconds") % 3600) / 60), 2, "0"),
        F.lit(":"),
        F.lpad(F.col("duration_seconds") % 60, 2, "0")
    )
).drop("ts_total_seconds", "duration_seconds")

场景2：new_ts是UNIX时间戳（毫秒/秒级）

如果new_ts是UNIX时间戳，直接基于时间戳计算差值：

from pyspark.sql import functions as F, Window as W

# 若new_ts是毫秒级，先转为秒级；如果是秒级可跳过此步
df_processed = df_subs_loc_movmnt.withColumn(
    "ts_seconds",
    F.col("new_ts") / 1000
)

window_spec = W.partitionBy("subs_no", "year")

df_result = df_processed.withColumn(
    "duration_seconds",
    F.col("ts_seconds") - F.min("ts_seconds").over(window_spec)
).withColumn(
    "duration",
    F.concat(
        F.lpad(F.floor(F.col("duration_seconds") / 3600), 2, "0"),
        F.lit(":"),
        F.lpad(F.floor((F.col("duration_seconds") % 3600) / 60), 2, "0"),
        F.lit(":"),
        F.lpad(F.col("duration_seconds") % 60, 2, "0")
    )
).drop("ts_seconds", "duration_seconds")

说明 #

• 通过将时间转换为数值类型（秒数），可以直接进行差值计算，避免字符串运算的错误。
• 使用lpad确保时分秒都是两位数字，格式统一为HH:mm:ss。

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