一个解决方案是准备间隔并进行连接：

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType
from pyspark.sql.functions import col, expr, to_timestamp
from datetime import datetime, timedelta
spark = SparkSession.builder.appName("example").getOrCreate()
# Create the DataFrame example:
schema = StructType([
    StructField("ID", StringType(), True),
    StructField("rh_start_time", TimestampType(), True),
    StructField("rh_end_time", TimestampType(), True),
    StructField("total_duration", IntegerType(), True)
])
def to_ts(date):
    return datetime.strptime(date, "%d-%m-%Y %H:%M:%S")
data = [
    ("5421833835", to_ts("31-12-2023 13:26:53"), to_ts("31-12-2023 13:27:03"), 10),
    ("5421833961", to_ts("31-12-2023 13:23:50"), to_ts("31-12-2023 13:39:10"), 360)
]
data = spark.createDataFrame(data, schema=schema)
data.show()
# Create all dates (if necessary, you can search the min and max in data):
start_date = datetime(2023, 12, 31)
end_date = datetime(2024, 1, 1)
interval = timedelta(minutes=15)
timestamps = [start_date + i * interval for i in range(int((end_date - start_date).total_seconds() // (15 * 60) + 1))]
raw_ts = [(timestamp,) for timestamp in timestamps]
column = "interval_start"
intervals = spark.createDataFrame(raw_ts, [column])
intervals = intervals.withColumn(column, col(column).cast(TimestampType()))
intervals.show()
# Do a join:
result = data.join(intervals, on=(
    (intervals["interval_start"] >= data["rh_start_time"] - expr("INTERVAL 15 MINUTES"))
    & (intervals["interval_start"] <= data["rh_end_time"])
))
result.show()

字符串
或者，您可以在开始时间的楼层上进行分解：

from pyspark.sql.functions import col, floor, explode
data.withColumn(
    'start_floor',
    (F.floor(F.col('rh_start_time').cast('integer') / (60 * 15)) * (60 * 15)).cast('timestamp')
).withColumn(
    "interval_start",
    F.explode(F.expr("sequence(start_floor, rh_end_time, interval 15 minutes)")),
).show()

型