apache-spark - 如何访问 mapGroupsWithState 中的 stateSnapshot 或在流之间共享 GroupState？

Question

通过 DStream API，可以使用 MapWithStateDStream.stateSnapshots() 访问有状态流的快照状态.在新的 Structured Streaming API 中，在我看来，只有传递给 mapGroupsWithState 的函数才能访问和更新状态。

我想根据输入事件创建流的内存中分布式状态。然后通过将另一个流中的事件加入第一个事件的(完整)状态来丰富事件，即不加入第一个流本身。

使用 DStream API，我只需将第二个流与第一个流的 stateSnapshot 连接起来。新的 Structured Streaming API 是否缺少此功能，或者是否有一种新的更好/更清洁的方式在两个流之间共享 GroupState？

Answer 1

Is this feature missing on the new Structured Streaming API or is there a new better/cleaner way of doing this?

没有stateSnapshot作为结构化流 (SS) 的一部分开箱即用。我假设这可以完成，也许在更高版本的 SS 中。我不确定它是否符合他们的设计目标，因为对于任意流的最终用户完全隐藏了状态，尽管这对于通过 (flat)mapGroupsWithState 使用自定义状态的人可能有用。 .

为了滚动你自己的“快照”，你总是可以输出你在GroupState[S]中的中间状态。对于生成的每个批处理，即:

def updateSessionEvents(
  id: Int,
  userEvents: Iterator[UserEvent],
  state: GroupState[UserSession]): Option[UserSession] = {
    // Do stuff
    val someState = ??? // update state
    someState
}

然后，您将一直处于整个状态。这意味着您现在需要维护一些标志，指示状态是否真正完成，以免将不完整的状态发送到下游，而这不是本意。