1

我想会话这个流：1,1,1,2,2,2,2,2,3,3,3,3,3,3,3,0,3,3,3,5, ...这些会议：

1,1,1
2,2,2,2,2
3,3,3,3,3,3,3
0
3,3,3
5

我编写了 CustomTrigger 来检测流元素何时从 1 变为 2（2 变为 3、3 变为 0 等等），然后触发触发器。但这不是解决方案，因为当我处理 2 的第一个元素并触发触发器时，窗口将为 [1,1,1,2] 但我需要在 1 的最后一个元素上触发触发器。

这是我的自定义触发器类中我的 onElement 函数的伪代码：

override def onElement(element: Session, timestamp: Long, window: W, ctx: TriggerContext): TriggerResult = {
    if (prevState == element.value) {
      prevState = element.value
      TriggerResult.CONTINUE
    } else {
      prevState = element.value
      TriggerResult.FIRE
    }
}

我怎么解决这个问题？

4

1 回答 1

2

我认为 a FlatMapFunctionwith aListState是实现这个用例的最简单方法。

当一个新元素到达时（即flatMap()方法被调用），你检查值是否改变了。如果值未更改，则将元素附加到状态。如果值更改，则将当前列表状态作为会话发出，清除列表，并将新元素作为第一个插入到列表状态中。

但是，您应该记住，这是假设元素的顺序保持不变。Flink 确保在一个分区内，即只要元素没有被打乱并且所有操作符以相同的并行度运行。

于 2017-06-19T07:57:40.453 回答