如何基于PySpark/Spark SQL拆分跨天时间区间并计算每日时长？

我之前在处理类似的跨天时间区间拆分需求时，踩过和你一样的坑——一开始尝试用concat+explode处理多列数组怎么都搞不定，后来找到了更简洁的方案，不管是用PySpark DataFrame API还是Spark SQL都能完美解决，下面给你详细拆解：

一、PySpark DataFrame API 方案 #

核心思路 #

1. 先计算每个时间区间覆盖的所有日期，用sequence函数生成连续的日期序列
2. 用explode把日期序列展开成单条记录，每条记录对应区间内的一天
3. 对每个日期，计算当天实际的起止时间：取原区间Start和当天0点的最大值作为当日Start，取原区间Stop和当天23:59:59.997的最小值作为当日Stop
4. 过滤掉无效记录（比如极端情况下当日Start > Stop的情况）

代码实现 #

首先创建测试数据模拟你的场景：

from pyspark.sql import functions as F

# 模拟你的数据集
data = [
    ("2021-02-02 18:00:00.000", "2021-02-03 02:00:00"),
    ("2021-02-02 18:00:00.000", "2021-02-08 02:00:00"),
    ("2021-02-05 09:00:00", "2021-02-05 17:00:00")  # 新增不跨天的测试用例
]

df = spark.createDataFrame(data, ["Start", "Stop"]) \
    .withColumn("Start", F.to_timestamp("Start")) \
    .withColumn("Stop", F.to_timestamp("Stop"))

然后执行拆分逻辑：

# 生成每个区间覆盖的日期序列并展开
df_with_dates = df.withColumn(
    "date_range",
    F.sequence(
        F.to_date("Start"),
        F.to_date("Stop"),
        F.expr("interval 1 day")
    )
).withColumn("current_date", F.explode("date_range"))

# 计算每日实际起止时间，并过滤无效记录
result_df = df_with_dates.withColumn(
    "daily_start",
    F.greatest("Start", F.to_timestamp(F.date_format("current_date", "yyyy-MM-dd 00:00:00.000")))
).withColumn(
    "daily_stop",
    F.least("Stop", F.to_timestamp(F.date_format("current_date", "yyyy-MM-dd 23:59:59.997")))
).filter(F.col("daily_start") <= F.col("daily_stop")) \
  .select("daily_start", "daily_stop") \
  .orderBy("daily_start")

# 查看结果
result_df.show(truncate=False)

扩展：计算每日时长 #

如果需要直接得到每日的时长，可以在结果中增加计算列：

result_df_with_duration = result_df.withColumn(
    "daily_duration_seconds",
    F.unix_timestamp("daily_stop") - F.unix_timestamp("daily_start")
).withColumn(
    "daily_duration_hours",
    F.round((F.unix_timestamp("daily_stop") - F.unix_timestamp("daily_start")) / 3600, 2)
)

result_df_with_duration.show(truncate=False)

二、Spark SQL 方案 #

如果更习惯用SQL处理，思路和上面完全一致，只是用SQL语法实现：

代码实现 #

先创建临时视图：

df.createOrReplaceTempView("time_intervals")

然后执行SQL查询：

WITH date_ranges AS (
    SELECT
        Start,
        Stop,
        explode(sequence(to_date(Start), to_date(Stop), interval 1 day)) AS current_date
    FROM time_intervals
)
SELECT
    greatest(Start, to_timestamp(date_format(current_date, 'yyyy-MM-dd 00:00:00.000'))) AS daily_start,
    least(Stop, to_timestamp(date_format(current_date, 'yyyy-MM-dd 23:59:59.997'))) AS daily_stop,
    -- 计算每日时长（秒和小时）
    unix_timestamp(least(Stop, to_timestamp(date_format(current_date, 'yyyy-MM-dd 23:59:59.997')))) - 
    unix_timestamp(greatest(Start, to_timestamp(date_format(current_date, 'yyyy-MM-dd 00:00:00.000')))) AS daily_duration_seconds,
    round(
        (unix_timestamp(least(Stop, to_timestamp(date_format(current_date, 'yyyy-MM-dd 23:59:59.997')))) - 
        unix_timestamp(greatest(Start, to_timestamp(date_format(current_date, 'yyyy-MM-dd 00:00:00.000')))))/3600, 2
    ) AS daily_duration_hours
FROM date_ranges
WHERE greatest(Start, to_timestamp(date_format(current_date, 'yyyy-MM-dd 00:00:00.000'))) <= 
      least(Stop, to_timestamp(date_format(current_date, 'yyyy-MM-dd 23:59:59.997')))
ORDER BY daily_start;

为什么之前的concat+explode方案没成功？ #

你提到尝试用concat和explode拼接多列数组后拆分失败，原因是explode只能单独展开一个数组列，如果要同时关联多列数据，需要把多列打包成结构体数组再展开，但这种方式比生成日期序列再处理要复杂得多。上面的方案通过先生成日期序列，再基于日期计算当日起止时间，逻辑更清晰，也更容易维护。

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