在flink全局窗口中重新处理未更改的元素，并进行连接转换

Question

Flink转换（连接）流中有一些元素正在重新处理，即使它们没有被修改。

假设我们有3个元素：1、2和3。插入它们时，会发生以下情况：

• 插入第一个元素1时，输出为：1
• 插入第二个元素2时，输出为：1-

在最后一次插入中，1或2没有任何更改，因此没有理由重新处理它们。

再处理规则：

• 只有同一出版商的书籍会被重新处理。这意味着当插入出版商2的书籍时，只有出版商2的书籍会被重新处理。我们的目标是不重新处理任何书籍，因为它们不受现有新书的影响。
• 当一个出版商被修改时，只有那个出版商的书被重新处理。（没关系）

联接后正在使用全局窗口，如下所示：

            bookStream
                .join(publisherStream)
                .where(book -> book.publisherId)
                .equalTo(publisher -> publisher.id)
                .window(GlobalWindows.create())
                .trigger(new ForeverTrigger<>())
                .apply(new JoinFunction<Book, Publisher, Book_Publisher>() {
                    @Override
                    public Book_Publisher join(Book book, Publisher publisher) throws Exception {
                        return new Book_Publisher(book, publisher);
                    }
                })

ForeverTrigger实施：

public class ForeverTrigger<T, E extends Window> extends Trigger<T, E> {

    @Override
    public TriggerResult onElement(T element, long timestamp, E window, TriggerContext ctx) throws Exception {
        return TriggerResult.FIRE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, E window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, E window, TriggerContext ctx) throws Exception {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(E window, TriggerContext ctx) throws Exception {}
}

对于这个用例，有必要始终存储所有元素，因为如果一本书被更新，我们需要有相应的出版商加入，反之亦然。因此，从bookStream或publisherStream中删除元素不是一个选项。

一个解决方案是使用TableAPI，如下所述：为什么Flink在DataStream join全局窗口上发出重复记录？。这将起作用，然后可以转换为数据流。然而，我希望避免将表API的使用与datastream API的使用混为一谈，特别是因为项目的主要目标是通用化和自动化flink管道的创建，这意味着要通用化的API将是两个而不是一个。因此，如果有一个不同的有效解决方案，那就太好了。

另一种解决方案是逐出或过滤元素，正如上面链接的同一篇文章中所提到的，但这似乎效率低下，因为它仍然需要处理元素，以便逐出/过滤它们。这需要保留以前状态的列表并比较传入元素。

理想情况下，Flink知道只处理包含更改的元素。是否有有效的解决方案来执行与数据流的连接并只处理修改后的元素？

Answer 1

窗口连接的设计并没有考虑到这种情况。为了有效地处理这个问题，我认为您需要在API堆栈中降低一个级别并使用KeyedCoProcessFunctions，或者提高一个级别并使用表API。