汀的知识碎片

07 flatMapGroupsWithState：任意有状态逻辑的终极武器

17 min read

07 flatMapGroupsWithState：任意有状态逻辑的终极武器

摘要

window() 聚合和 groupBy().agg() 能处理大多数时间窗口统计需求，但面对自定义状态机（如用户行为序列检测、欺诈检测、会话状态管理）时力不从心——这些场景需要对每个 Key 维护任意复杂的状态对象，并在每次接收新数据时执行自定义的状态转移逻辑。flatMapGroupsWithState（以及其简化版 mapGroupsWithState）是 Structured Streaming 为此提供的”逃生舱”：它给用户完全的状态控制权，可以读取和更新任意 Scala/Java 对象作为状态，并在每次有新数据时执行用户定义的函数。本文深度讲解 GroupState 接口的读写语义、两种超时机制（ProcessingTime Timeout vs EventTime Timeout）的底层原理与使用场景、flatMapGroupsWithState 与 mapGroupsWithState 的区别，以及用户行为序列检测、欺诈检测等典型生产案例的完整实现。

第 1 章 为什么需要自定义状态

1.1 内置聚合算子的边界

Structured Streaming 的内置有状态算子（window().groupBy().agg()、dropDuplicates()）能处理”数值聚合”类需求：

# 内置聚合擅长的：数值统计
events.groupBy(window("ts", "5 min"), "category") \
      .agg(sum("amount"), count("*"))

但有一类需求是内置聚合无法表达的：状态的转移取决于历史事件的序列。

典型场景一：欺诈检测（规则引擎）

检测规则：同一用户在 10 分钟内，先后出现”小额试探”（< 10 元）和”大额消费”（> 1000 元），标记为疑似欺诈。

这个规则需要：

  1. 记住该用户最近是否有过”小额试探”事件（及其时间）
  2. 当收到新的”大额消费”事件时，检查是否在 10 分钟内有历史的”小额试探”
  3. 满足条件则输出告警，并清除状态

内置的 agg() 只能做累加/计数，无法表达”先有小额再有大额”这种有序的状态转移逻辑。

典型场景二：用户会话分析（行为路径）

分析每个用户的完整会话路径（首页 → 商品详情 → 加购 → 下单），统计转化率和路径分布。需要按用户记录当前会话内的所有事件序列。

典型场景三：机器状态监控

监控 IoT 设备的状态变化（正常 → 预警 → 告警 → 恢复）。状态转移规则复杂，需要按设备 ID 维护有限状态机。

这些场景的共同特点：状态是任意复杂的业务对象，状态转移逻辑由业务规则决定——既不是简单求和，也不是窗口聚合，而是一段完整的业务逻辑代码。flatMapGroupsWithState 正是为此设计的。

第 2 章 mapGroupsWithState vs flatMapGroupsWithState

2.1 两者的核心区别

维度mapGroupsWithStateflatMapGroupsWithState
每次调用的输出行数恰好输出 1 行输出 0 到多行（通过迭代器）
适用场景每次有新数据时必须输出一行结果只在特定条件下输出（如检测到异常才告警）
超时处理超时时调用用户函数，输出 1 行超时时调用用户函数，可输出 0 行（不告警）或多行
// mapGroupsWithState：每次调用必须返回 1 行（适合实时仪表盘类需求）
def updateUserMetrics(
    userId: String,
    events: Iterator[Event],
    state: GroupState[UserMetricsState]
): UserMetrics = {
    // 更新状态
    val current = state.getOption.getOrElse(UserMetricsState())
    val updated = events.foldLeft(current)(_.update(_))
    state.update(updated)
    // 必须返回 1 行
    UserMetrics(userId, updated.totalAmount, updated.eventCount)
}
 
// flatMapGroupsWithState：可返回 0 到多行（适合告警、异常检测）
def detectFraud(
    userId: String,
    events: Iterator[Event],
    state: GroupState[FraudDetectionState]
): Iterator[FraudAlert] = {
    // 更新状态
    // 只在检测到欺诈时才输出告警行（通常大多数 Key 返回空 Iterator）
    if (fraudDetected) Iterator(FraudAlert(userId, ...))
    else Iterator.empty  // 大多数用户无异常，输出 0 行
}

2.2 选择原则

  • 需要每个 Key 在每批次都有输出（如实时更新的仪表盘指标）→ mapGroupsWithState
  • 只在特定条件触发输出（如告警、异常检测）→ flatMapGroupsWithState（大多数 Key 返回空）
  • 需要一次输出多行（如从一个会话状态中展开多条记录）→ flatMapGroupsWithState

第 3 章 GroupState 接口深度解析

3.1 GroupState 的读写操作

GroupState[S] 是用户与 State Store 交互的唯一接口，提供以下核心操作：

trait GroupState[S] {
    // 读取状态
    def exists: Boolean                     // 该 Key 的状态是否存在
    def get: S                              // 读取状态（不存在时抛出 NoSuchElementException）
    def getOption: Option[S]                // 安全读取（不存在时返回 None）
    
    // 写入状态
    def update(newState: S): Unit           // 更新状态（不存在则创建）
    def remove(): Unit                      // 删除状态（主动清理，释放 State Store 空间）
    
    // 超时控制
    def setTimeoutDuration(duration: Duration): Unit    // 设置 ProcessingTime 超时
    def setTimeoutTimestamp(timestamp: Long): Unit      // 设置 EventTime 超时时间戳
    def setTimeoutTimestamp(timestamp: java.sql.Timestamp): Unit
    
    // 上下文信息
    def isTimingOut: Boolean                // 当前调用是否由超时触发（不是新数据触发）
    def getCurrentWatermarkMs(): Long       // 当前 Watermark（毫秒，用于 EventTime 超时）
    def getCurrentProcessingTimeMs(): Long  // 当前处理时间（毫秒，用于 ProcessingTime 超时）
    
    // 批次信息
    def hasTimedOut: Boolean               // 同 isTimingOut（别名）
}

3.2 典型使用模式：欺诈检测完整实现

import org.apache.spark.sql.streaming.{GroupState, GroupStateTimeout, OutputMode}
import java.sql.Timestamp
 
// 状态对象（可序列化的 case class）
case class FraudState(
    smallTxnTime: Option[Long],   // 最近一次小额试探的时间戳（毫秒）
    smallTxnAmount: Double        // 小额试探的金额
)
 
// 输入事件
case class Transaction(userId: String, amount: Double, ts: Long)
 
// 输出告警
case class FraudAlert(userId: String, reason: String, alertTime: Long)
 
def detectFraud(
    userId: String,
    events: Iterator[Transaction],
    state: GroupState[FraudState]
): Iterator[FraudAlert] = {
    
    val alerts = scala.collection.mutable.ArrayBuffer[FraudAlert]()
    
    // 超时触发：10 分钟无新事件，清理状态（该用户无后续操作）
    if (state.hasTimedOut) {
        state.remove()
        return Iterator.empty
    }
    
    // 读取当前状态（首次无状态）
    var currentState = state.getOption.getOrElse(FraudState(None, 0.0))
    
    // 处理本批次的所有事件（按时间排序）
    val sortedEvents = events.toSeq.sortBy(_.ts)
    
    for (event <- sortedEvents) {
        if (event.amount < 10.0) {
            // 小额试探：记录到状态
            currentState = FraudState(Some(event.ts), event.amount)
        } else if (event.amount > 1000.0 && currentState.smallTxnTime.isDefined) {
            // 大额消费：检查是否在 10 分钟内有小额试探
            val smallTxnAge = event.ts - currentState.smallTxnTime.get
            if (smallTxnAge <= 10 * 60 * 1000L) {  // 10 分钟内
                // 检测到欺诈！
                alerts += FraudAlert(
                    userId,
                    s"Small txn ${currentState.smallTxnAmount} followed by large txn ${event.amount} within 10 minutes",
                    event.ts
                )
                currentState = FraudState(None, 0.0)  // 清除状态，避免重复告警
            }
        }
    }
    
    // 更新状态，并设置 10 分钟的处理时间超时（10 分钟无新事件自动清理状态）
    if (currentState.smallTxnTime.isDefined) {
        state.update(currentState)
        state.setTimeoutDuration("10 minutes")  // ProcessingTime 超时
    } else {
        state.remove()  // 无待检测状态，主动清理
    }
    
    alerts.iterator
}
 
// 使用 flatMapGroupsWithState
val alerts = transactions
    .as[Transaction]
    .groupByKey(_.userId)
    .flatMapGroupsWithState(
        outputMode = OutputMode.Append(),
        timeoutConf = GroupStateTimeout.ProcessingTimeTimeout
    )(detectFraud)

第 4 章 超时机制：状态的自动清理

4.1 为什么需要超时

有状态算子面临一个长期运行问题：随着时间推移，越来越多的 Key 积累在 State Store 中，即使某些 Key 的用户已经长时间没有新事件（用户流失、设备下线）。如果不主动清理这些”僵尸状态”，State Store 会无限增长。

超时机制（Timeout）允许用户指定：如果某个 Key 超过一定时间没有新数据，就触发一次用户函数调用（state.hasTimedOut == true），让用户决定是否删除状态。

4.2 ProcessingTime Timeout

ProcessingTime 超时：基于系统时钟（墙钟时间），从最后一次 setTimeoutDuration 调用起计时，超过指定时长后触发。

// 在函数中设置超时
state.setTimeoutDuration("10 minutes")
// 等价于：state.setTimeoutDuration(10 * 60 * 1000L)
 
// 超时触发后
if (state.hasTimedOut) {
    val expiredState = state.get
    state.remove()
    // 可以输出一条"会话结束"记录
    return Iterator(SessionEnd(key, expiredState))
}

ProcessingTime 超时的特点：

  • 与数据流无关：即使没有该 Key 的新数据到来，超时也会在挂钟时间到达后触发
  • 精度约为一个批次间隔：Spark 在每个批次结束时检查超时，而不是实时检查；精度 ≈ processingTime 触发间隔
  • 不依赖 Watermark：不需要设置 withWatermark，任何查询都可以使用

适用场景：

  • 用户会话超时（30 分钟无操作结束会话）
  • 状态内存保护（防止长时间未活跃的 Key 一直占用内存）
  • 不依赖事件时间的业务逻辑

4.3 EventTime Timeout

EventTime 超时：基于 Watermark（事件时间进度），当 Watermark 超过为某个 Key 设置的超时时间戳时触发。

// 需要先定义 Watermark
val df = events.withWatermark("event_time", "10 minutes")
 
// 在函数中设置超时时间戳（基于事件时间）
state.setTimeoutTimestamp(
    state.getCurrentWatermarkMs() + 30 * 60 * 1000L  // Watermark + 30 分钟后超时
)
 
// 或者基于最后一次事件的时间
state.setTimeoutTimestamp(lastEventTime + 30 * 60 * 1000L)

EventTime 超时的触发条件：

Key 的超时时间戳 = T
当前 Watermark = W

触发条件：W > T
→ 当 Watermark 推进到超过 T 时，该 Key 触发超时

EventTime 超时的特点：

  • 语义与事件时间对齐：超时基于数据的时间逻辑，不受系统时钟影响
  • 依赖 Watermark 推进：如果 Watermark 停滞（见第 04 篇），超时也不会触发
  • 可重放性：相同的数据序列总是产生相同的超时触发，利于调试和测试

适用场景：

  • 基于事件时间的会话超时（如”最后一个事件的事件时间 + 30 分钟后会话结束”）
  • 需要可重放语义（批处理补跑历史数据时行为一致）

4.4 两种超时的选择对比

维度ProcessingTime 超时EventTime 超时
时间参考系统挂钟时间Watermark（事件时间进度）
Watermark 依赖不需要需要 withWatermark
历史数据重放超时行为不一致（取决于实际运行时间）超时行为与数据一致（可重放）
Watermark 停滞时超时正常触发超时不触发（Watermark 不推进）
精度≈ 批次间隔≈ 批次间隔
典型场景实时会话管理、内存保护事件时间语义的会话分析

第 5 章 性能考量与生产注意事项

5.1 状态对象的序列化

GroupState[S] 中的状态对象 S 需要在 State Store（通常是 RocksDB 或 HDFS）中序列化存储。Spark 使用 Java 序列化 或 Kryo 序列化状态对象。

性能优化建议：

  • 状态对象应该尽量小（只存必要字段）
  • 使用 Kryo 序列化（比 Java 序列化快 3-10 倍）：spark.serializer=org.apache.spark.serializer.KryoSerializer
  • 避免在状态中存储大对象（如 List[String] 存储所有历史事件），改为只存统计值

5.2 输出模式的限制

flatMapGroupsWithState 和 mapGroupsWithState 只支持两种输出模式：

  • Append 模式（仅限 flatMapGroupsWithState）：只在确定输出时才写出；超时触发时可以输出”状态结束”记录
  • Update 模式：每批次输出有变化的 Key 的结果

不支持 Complete 模式（Complete 需要全量状态快照，与自定义状态管理冲突）。

生产避坑

状态更新函数（用户定义的函数）在 Spark Task 中执行，必须是可序列化的（不能捕获非序列化的外部对象，如数据库连接、文件句柄）。如果需要访问外部资源（如查询 Redis），应在函数内部创建连接，使用完后关闭，不要在 Driver 端创建后传入 Task。

小结

flatMapGroupsWithState 是 Structured Streaming 最灵活的有状态算子：

  • 核心价值：用户完全掌控状态读写和转移逻辑，可以实现任意复杂的业务状态机
  • GroupState 接口：get/update/remove 管理状态存储；setTimeoutDuration/setTimeoutTimestamp 设置自动超时清理
  • ProcessingTime 超时：基于系统时钟，不依赖 Watermark，精度约为批次间隔，适合实时会话管理
  • EventTime 超时：基于 Watermark，语义与事件时间对齐，可重放，依赖 Watermark 正常推进
  • 欺诈检测模式：hasTimedOut 处理超时清理，foldLeft 按时序处理一批事件，主动 remove() 避免状态无限增长

第 08 篇深入流-流 Join：两条流的时间对齐语义、State Store 中的数据缓冲机制、Watermark 如何控制 Join Buffer 的清理，以及支持和不支持的 Join 类型。

思考题

  1. flatMapGroupsWithState 允许用户完全自定义状态管理逻辑，是最灵活的有状态算子。但这种灵活性也带来了风险——用户可能在状态中存储无限增长的数据结构（如 List）。与内置的窗口聚合算子相比，flatMapGroupsWithState 的状态生命周期管理完全由用户负责。如果用户忘记在状态函数中清理过期状态，State Store 会如何响应？State TTL（GroupStateTimeout）与 Watermark-based 清理有什么区别？
  2. flatMapGroupsWithState 使用 GroupStateTimeout 来处理超时逻辑（如用户长时间无活动则关闭 Session）。ProcessingTimeTimeout 和 EventTimeTimeout 的触发时机和语义有什么不同？在事件时间乱序严重的场景下，EventTimeTimeout 是否比 ProcessingTimeTimeout 更安全？
  3. flatMapGroupsWithState 要求用户定义的状态类型必须是可序列化的（Encoder 支持的类型）。在作业升级时，如果新版本的状态类型与旧版本不兼容（比如增加了字段），Spark 从旧 Checkpoint 恢复时会发生什么？有哪些工程实践可以实现状态 Schema 的平滑演进？

参考资料