Apache Flink连续拆分奇怪的行为

Question

我相信，当连续完成两个拆分时，Flink的行为很奇怪。我可能在我的实现逻辑中有一些错误，这就是为什么我在这里发帖征求您的意见。

最小示例：我有一个包含单词Apple、Banana和Orange的文本文件。我将其作为源在流执行环境中传递。我进行了第一次拆分，其中选择条件是参数是否为单词“Apple”。如果是，我将其放在“主题”Apples中，否则放在“主题”NotApples中。然后我在此拆分流中选择“主题”NotApples并再次拆分它，但这次条件检查参数是否为单词“Orange”。如果是，则将其放置在“主题”橙子中，否则放置在“主题”NotOrang中。

最后，当我打印最后一条分割流的主题NotOranges时，我所期望的是只打印“香蕉”一词。然而，我实际上打印的是“苹果”和“香蕉”两个词。我注意到，在进行第二次拆分时，处理的流不是仅包含我从中选择的主题元素（即NotApples）的流，而是包含所有元素的流。我错过什么了吗？

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> datastream = env.readTextFile("input.txt");
SplitStream<String> splitStream1 = datastream.split(new OutputSelector<String>() {

    @Override
    public Iterable<String> select(String arg0) {
        List<String> output = new ArrayList<String>();
        if (arg0.equals("Apple")) {
            output.add("Apples");
        } else {
            output.add("NotApples");
        }
        return output;
    }
});


DataStream<String> notApplesStream = splitStream1.select("NotApples");
SplitStream<String> splitStream2 = notApplesStream.split(new OutputSelector<String>() {

    @Override
    public Iterable<String> select(String arg0) {
        List<String> output = new ArrayList<String>();
        if (arg0.equals("Orange")) {
            output.add("Oranges");
        } else {
            output.add("NotOranges");
        }
        return output;
    }
});

DataStream<String> notApplesAndNotOrangesStream = splitStream2.select("NotOranges");
notApplesAndNotOrangesStream.print();
env.execute("SplitTest");

输出：

1> Apple
1> Apple
1> Banana
2> Apple
2> Apple
2> Apple
4> Apple
4> Apple
4> Banana
3> Apple
3> Banana
3> Apple

注意：我知道我可以用一个拆分来实现相同的逻辑（在这个逻辑中，我检查参数是“Apple”还是“Organge”）。然而，这不是我问题的重点。我最初在我编写的一个更复杂的程序中注意到这种行为，其中需要两个结果拆分，所以我决定尝试在一个最小的示例中重新创建它，以检查我是否可以重现它。

Answer 1

考虑到我对拆分/选择是如何实现的所知，如果这不起作用，我也不会感到惊讶（尽管我还不太清楚）。此外，拆分/选择最近已被弃用（尽管尚不清楚它是否真的会消失）。

进行拆分/选择的更好方法是通过侧面输出。这是一种更强大的机制，具有更清晰的实现。

Answer 2

最近在邮件列表上讨论了这种不正确的行为，主题是“关于为DataStream API弃用拆分/选择”。我认为关键评论是：

首先，我们必须承认当前拆分/选择的实现是有缺陷的。我粗略地浏览了源代码，问题可能是对于连续的选择/拆分，前一个将在StreamGraph生成阶段被后一个覆盖。这就是为什么我们在FLINK-11084中禁止这种连续逻辑。

在查看了FLINK-11084和由此产生的补丁之后，我相信如果您连续进行两次拆分/选择，最近发布的Flink将引发异常。