Spark 2.2 Structured Streaming中如何获取各分组最新条目？

阿华AIGC实验室

2026-5-21

解决方案：Spark 2.2 Structured Streaming 获取分组最新条目

嘿，针对你在Spark 2.2 Structured Streaming里碰到的分组取最新条目的问题——没法做流DF的Join（包括自连接）、也不能直接用排序函数，结合你补充的“数据已按时间排序”的前提，我整理了两个可行的方案：

方案一：基于预排序数据的`groupBy + agg(last())`

如果你的流数据已经按时间戳在全局或分区内严格升序排列，那直接用分组后聚合取last()的方式就能拿到每个分组的最新记录，简单高效，不用搞复杂的状态管理。

代码示例：

假设你的流DF有group_id（分组键）、event_time（时间戳）、data（业务数据）这几个字段：

import org.apache.spark.sql.functions._

// 读取流数据（这里以Kafka为例，你可以换成自己的数据源）
val streamingDF = spark.readStream
  .format("kafka")
  .load()
  .selectExpr("cast(value as string) as json")
  .select(from_json($"json", yourSchema).as("data"))
  .select("data.group_id", "data.event_time", "data.*")

// 分组后取每个分组的最后一条（也就是最新的）记录
val latestPerGroupDF = streamingDF
  .groupBy($"group_id")
  .agg(
    last($"event_time").as("latest_event_time"),
    last($"data").as("latest_data")
    // 其他需要保留的字段，都用last()来聚合就行
  )

// 输出结果到控制台（按需替换输出方式）
latestPerGroupDF.writeStream
  .outputMode("update")
  .format("console")
  .start()
  .awaitTermination()

注意事项：

一定要保证数据在分组前已经按event_time升序排好，不然last()可能拿不到真正的最新记录；

建议加上水印（withWatermark）来清理过期的分组状态，避免内存爆掉：

val streamingDFWithWatermark = streamingDF
  .withWatermark("event_time", "1 hour") // 时间根据你的业务场景调
  .groupBy($"group_id")
  .agg(...)

方案二：用`mapGroupsWithState`自定义状态管理

如果数据没预排序，或者你需要更灵活的判断逻辑（比如靠时间戳对比，而不是记录位置），那可以用Spark 2.2刚引入的mapGroupsWithState API，手动维护每个分组的最新条目，这也是2.2版本里处理这类问题的硬核方法。

代码示例：

先定义好业务数据和状态的样例类，再写状态更新逻辑：

import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout}

// 业务数据的样例类，根据你的实际字段调整
case class Event(groupId: String, eventTime: Long, data: String)
// 用来存储每个分组最新记录的状态类
case class LatestEvent(groupId: String, latestTime: Long, latestData: String)

// 核心的状态更新函数：遍历每个批次的事件，对比时间戳保留最新的
def updateLatestEvent(
  groupId: String,
  events: Iterator[Event],
  state: GroupState[LatestEvent]
): Iterator[LatestEvent] = {
  // 先拿当前状态里已有的最新记录（如果有的话）
  val currentLatest = state.getOption()

  // 遍历当前批次的所有事件，找出比状态里更新的记录
  val newLatest = events.foldLeft(currentLatest) {
    case (Some(existing), event) =>
      if (event.eventTime > existing.latestTime) {
        Some(LatestEvent(groupId, event.eventTime, event.data))
      } else {
        Some(existing)
      }
    case (None, event) =>
      Some(LatestEvent(groupId, event.eventTime, event.data))
  }

  // 更新状态：有新的最新记录就更新，没有就删掉状态
  newLatest match {
    case Some(event) => state.update(event)
    case None => state.remove()
  }

  // 返回最终的最新记录
  newLatest.toIterator
}

// 读取流数据并应用状态更新逻辑
val streamingDF = spark.readStream
  .format("kafka")
  .load()
  .selectExpr("cast(value as string) as json")
  .select(from_json($"json", schemaOf[Event]).as("event"))
  .select("event.*")

val latestPerGroupDF = streamingDF
  .groupByKey(_.groupId)
  .mapGroupsWithState(GroupStateTimeout.ProcessingTimeTimeout())(updateLatestEvent)

// 输出结果
latestPerGroupDF.writeStream
  .outputMode("update")
  .format("console")
  .start()
  .awaitTermination()