如何调试键控窗口上的多个触发事件

Question

我的数据流来自一个定制的SourceFunction，它以确定性的顺序发出窗口大小的字符串序列。其目的是基于EventTime在keyedstream上创建滑动窗口，以便对累积的字符串进行处理。为了分配EventTime和水印，我将一个带有PeriodicWatermarks的赋值器附加到流。滑动窗口使用自定义ProcessWindowFunction进行处理。

env.setStreamTimeCharacteristic(EventTime)
val seqStream = env.addSource(Seqstream)
    .assignTimestampsAndWatermarks(SeqTimeStampExtractor())
    .keyBy(getEventtimeKey)
    .window(SlidingEventTimeWindows.of(Time.milliseconds(windowSize), Time.milliseconds(slideSize)))

val result = seqStream.process(ProcessSeqWindow(target1))

My AssignerWithPeriodicWaterMarks 看起来像这样：

class FASTATimeStampExtractor : AssignerWithPeriodicWatermarks<FASTAstring> {
    var waterMark  = 9999L
    override fun extractTimestamp(element: FASTAstring, previousElementTimestamp: Long): Long {
        return element.f1
    }

    override fun getCurrentWatermark(): Watermark? {
        waterMark += 1
        return Watermark(waterMark)
    }
}

换句话说，源发出的每个元素都应该有自己的EventTime，并且应该发出水印，在该时间内不允许发生其他事件。在调试器中单步遍历流，表示生成了预期的EventTime/WatComments。

我的期望是ProcessSeqWindow.run（）应该在EventTime上使用与时间窗口成比例的多个元素调用（例如10 ms）。然而，我观察到的是run（）使用单个元素多次调用，并且相对于EventTime以任意顺序调用。当我将并行性强制为1时，行为仍然存在。我的问题是这是否可能是由每个窗口上的多个触发事件引起的，还是有其他可能的解释？我如何调试原因？

谢啦

Answer 1

水印在您的作业中的作用将是触发滑动事件时间窗口的关闭。为了正确发挥该作用，它们应该基于事件中的时间戳，而不是某个任意常量（9999L）。同一个对象负责提取时间戳并提供水印的原因是为了使该对象可以根据其对事件流中时间戳的观察来创建水印。因此，除非您的事件时间戳也基于递增类似的计数器，否则这可能会解释您看到的一些行为。

另一个问题是，虽然为每个事件调用extractTimestamp，但在定期水印分配器中，getCurrentWatermark方法每隔200毫秒(默认情况下)在单独的线程中调用一次。如果您希望在每个事件后都有水印，您将需要使用一个带有标点水印的assigner，尽管这样做有点反模式(因为有那么多水印会增加开销)。

如果你的时间戳是完全人为的，你可能会发现SlipingCountWindow更适合你正在做的事情。