Flink流窗口触发器

Question

我有flink stream，我在某个时间窗口上计算一些事情，比如说30秒。

这里发生的事情是，它给我的结果，我以前的窗口聚合以及。

假设前30秒我得到结果10。

接下来的三十秒我想要新的结果，而不是我得到最后一个窗口的结果新的等等。

因此，我的问题是如何为每个窗口获得新的结果。

Answer 1

我在这个问题上有点晚了，但是我在OP上遇到了同样的问题。我后来发现我自己的代码中有一个bug。仅供参考，我的错误可以作为解决您问题的参考。

// Old code (modified to be an example):
val tenSecondGrouping: DataStream[MyCustomGrouping] = userIdsStream
      .keyBy(_.somePartitionedKey)
      .window(TumblingProcessingTimeWindows.of(Time.of(10, TimeUnit.SECONDS)))
      .trigger(ProcessingTimeTrigger.create())
      .aggregate(new MyCustomAggregateFunc(new MyCustomGrouping()))

错误发生在新的MyCustomGrouping上：我无意中创建了一个singleton MyCustomGrouping对象，并在MyCustomAggregateFunc中重用它。随着更多滚动窗口的创建，后期的聚合结果变得疯狂！修复方法是在每次触发MyCustomAggregateFunc时创建新的MyCustomGrouping。因此：

// New code, problem solved
          ...
          .aggregate(new MyCustomAggregateFunc(() => new MyCustomGrouping())) 
// passing in a func to create new object per trigger

Answer 2

您描述的功能可以在翻滚窗口中找到：https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/windows.html#tumbling-windows

更详细一点和/或代码会有所帮助：）

Answer 3

您需要使用清除触发器。你想要的是FIRE_AND_PURGE（发射和删除窗口内容），默认的flink触发器所做的是FIRE（发射和保留窗口内容）。

input
    .keyBy(...)
    .timeWindow(Time.seconds(30))

    // The important part: Replace the default non-purging ProcessingTimeTrigger
    .trigger(new PurgingTrigger[..., TimeWindow](ProcessingTimeTrigger))

    .reduce(...)

要获得更深入的解释，请查看触发器和FIRE与FIRE_和_PURGE的对比。

触发器确定窗口（由窗口赋值器形成）何时准备好由窗口函数处理。每个WindowAssigner都带有一个默认触发器。如果默认触发器不符合您的需要，可以使用触发器（…）指定自定义触发器。

当触发器触发时，它可以触发或FIRE_AND_PURGE。当FIRE保留窗口内容时，FIRE_AND_PURGE删除其内容。默认情况下，预实现的触发器只是FIRE而不清除窗口状态。