如何合并Spark DataSet中满足时间间隔条件的Track数据？

阿华AIGC实验室

2026-5-25

轨迹融合（5分钟间隔合并）实现方案

这个轨迹融合需求本质是**会话划分（Sessionization）**的典型场景——把时间间隔不超过5分钟的连续轨迹归为同一组，再合并每组的轨迹信息。下面基于Spark Dataset给出具体实现步骤，适配你提到的Track.class场景（以Scala为例）：

1. 数据预处理：转换时间格式

首先需要把字符串类型的时间转成Spark的Timestamp类型，方便后续计算时间差：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.TimestampType
import org.apache.spark.sql.expressions.Window

// 假设你的原始Dataset是trackDs，类型为Dataset[Track]
val timestampTrackDs = trackDs
  .withColumn("start_ts", to_timestamp(col("start_time"), "HH:mm:ss"))
  .withColumn("end_ts", to_timestamp(col("end_time"), "HH:mm:ss"))

注意：如果你的时间包含日期（跨天场景），把格式串改成yyyy-MM-dd HH:mm:ss即可。

2. 计算轨迹间的时间间隔

用窗口函数lag获取上一条轨迹的结束时间，然后计算当前轨迹开始时间与上一条结束时间的时间差（单位：分钟）：

// 按开始时间排序的窗口
val timeWindow = Window.orderBy("start_ts")

val withIntervalDs = timestampTrackDs
  .withColumn("prev_end_ts", lag(col("end_ts"), 1).over(timeWindow))
  // 计算时间差：如果是第一条轨迹，时间差设为0；否则转成分钟
  .withColumn("interval_min", when(col("prev_end_ts").isNull, lit(0))
    .otherwise((col("start_ts").cast("long") - col("prev_end_ts").cast("long")) / 60))

3. 生成融合分组标识

当两条轨迹的时间间隔超过5分钟时，就新建一个分组；否则归为同一组。用累加求和的方式生成分组ID：

val withGroupDs = withIntervalDs
  .withColumn("group_id", sum(when(col("interval_min") > 5, 1).otherwise(0))
    .over(timeWindow.rangeBetween(Window.unboundedPreceding, 0)))

原理：每次遇到间隔>5分钟的轨迹，就给分组ID加1，这样连续的轨迹会共享同一个分组ID。

4. 分组合并轨迹信息

最后按分组ID聚合，合并每组内的轨迹：

val mergedTracks = withGroupDs
  .groupBy("group_id")
  .agg(
    // 合并所有关联的trackId，用逗号分隔
    concat_ws(",", collect_list("trackId")).alias("merged_track_ids"),
    // 取分组内最早的开始时间
    min("start_ts").alias("merged_start_time"),
    // 取分组内最晚的结束时间
    max("end_ts").alias("merged_end_time")
  )
  // 把时间转回到字符串格式（可选）
  .select(
    col("merged_track_ids"),
    date_format(col("merged_start_time"), "HH:mm:ss").alias("start_time"),
    date_format(col("merged_end_time"), "HH:mm:ss").alias("end_time")
  )