持久化Apache Flink窗口

Question

我正试图使用Flink以流媒体的方式使用消息队列中的有界数据。数据格式如下：

{"id":-1,"name":"Start"}
{"id":1,"name":"Foo 1"}
{"id":2,"name":"Foo 2"}
{"id":3,"name":"Foo 3"}
{"id":4,"name":"Foo 4"}
{"id":5,"name":"Foo 5"}
...
{"id":-2,"name":"End"}

可以使用事件ID确定消息的开始和结束。我想接收此类批次并将最新的（通过覆盖）批次存储在磁盘或内存中。我可以编写自定义窗口触发器来使用开始和结束标志提取事件，如下所示：

DataStream<Foo> fooDataStream = ...
AllWindowedStream<Foo, GlobalWindow> fooWindow = fooDataStream.windowAll(GlobalWindows.create())
.trigger(new CustomTrigger<>())
.evictor(new Evictor<Foo, GlobalWindow>() {
    @Override
    public void evictBefore(Iterable<TimestampedValue<Foo>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {
        for (Iterator<TimestampedValue<Foo>> iterator = elements.iterator();
             iterator.hasNext(); ) {
            TimestampedValue<Foo> foo = iterator.next();
            if (foo.getValue().getId() < 0) {
                iterator.remove();
            }
        }
    }

    @Override
    public void evictAfter(Iterable<TimestampedValue<Foo>> elements, int size, GlobalWindow window, EvictorContext evictorContext) {

    }
});

但是如何保持最新窗口的输出。一种方法是使用ProcessAllWindowFunction接收所有事件并手动将其写入磁盘，但这感觉像是一种黑客行为。我还研究了带有Flink CEP模式的表API（如此问题），但找不到在每个批处理之后清除表以丢弃前一批中的事件的方法。

Answer 1

有几件事阻碍了你的愿望：

（1） Flink的窗口操作符生成附加流，而不是更新流。它们不是为更新以前发出的结果而设计的。CEP也不会生成更新流。

（2） Flink的文件系统抽象不支持覆盖文件。这是因为对象存储（如S3）不太支持此操作。

我认为你的选择是：

（1）重做您的作业，使其产生更新（变更日志）流。您可以使用toChangelogStream或使用创建更新流的Table/SQL操作来执行此操作，例如GROUP BY（当它在没有时间窗口的情况下使用时）。最重要的是，您需要选择支持撤回/更新的接收器，例如数据库。

（2） 坚持生成附加流，并使用类似于FileSink的东西将结果写入一系列滚动文件。然后在Flink之外编写一些脚本，以获得您想要的结果。