我想我有办法解决你的问题：
给予 DataStream 的 (id, state, time) 作为：

val stateUpdates: DataStream[(Long, Int, ts)] = ???

实际状态更改如下所示：

val stateCntUpdates: DataStream[(Int, Int)] = s // (state, cntUpdate)
  .keyBy(_._1) // key by id
  .flatMap(new StateUpdater)
``` `StateUpdater` 是有状态的 `FlatMapFunction` . 它有一个键控状态，存储每个id的最后一个状态。对于每个输入记录，它返回两个状态计数更新记录： `(oldState, -1)` ,  `(newState, +1)` . 这个 `(oldState, -1)` 记录确保减少以前状态的计数。
接下来，根据每个状态和窗口聚合状态计数更改：

val cntUpdatesPerWindow: DataStream[(Int, Int, Long)] = stateCntUpdates // (state, cntUpdate, time)
.keyBy(_._1) // key by state
.timeWindow(Time.minutes(10)) // window should be non-overlapping, e.g. Tumbling
.apply(new SumReducer(), new YourWindowFunction())
``` SumReducer 对CNT更新和 YourWindowFunction 指定窗口的时间戳。此步骤聚合窗口中每个状态的所有状态更改。
最后，我们使用计数更新来调整当前计数。

val stateCnts: DataStream[(Int, Int, Long)] = cntUpdatesPerWindow // (state, count, time)
  .keyBy(_._1) // key by state again
  .map(new CountUpdater)
``` `CountUpdater` 是有状态的 `MapFunction` . 它有一个键控状态，存储每个状态的当前计数。对于每个传入的记录，将调整计数并创建一个记录 `(state, newCount, time)` 发射。
现在您有了一个流，每个状态都有新的计数（每个状态一个记录）。如果可能，您可以使用这些记录更新elasticsearch索引。如果需要收集给定时间内的所有状态计数，可以使用窗口来完成。
请注意：此程序的状态大小取决于唯一ID的数量。如果id空间增长非常快，可能会导致问题。