java - 如何提高 Flink 中数据流实现的不同计数？

Question

我在 Flink (Java) 中有这个程序，它可以计算数据流中的不同单词。我使用计数单词的示例来实现，并在同一时间应用另一个窗口来评估不同的值。该程序运行良好。但是，我担心我正在使用两个窗口来处理不同的计数。第一个窗口计算单词数，第二个窗口将单词数切换为 1，并将单词切换为 Tuple2 的第二个元素。我数了数 key 的数量。这是我的程序的输入和输出:

// input:
aaa
aaa
bbb
ccc
bbb
aaa
output:
(3,bbb-ccc-aaa)

如果我删除第二个窗口，它会显示每个键的所有评估并保存前一个窗口的状态。

// input:
aaa
aaa
bbb
ccc
bbb
aaa
// output:
3> (1,bbb)
3> (2,bbb-aaa)
3> (3,bbb-aaa-ccc)
// wait the first window to be evaluated.
// input:
aaa
aaa
bbb
ccc
bbb
aaa
// output:
3> (4,bbb-aaa-ccc-ccc)
3> (5,bbb-aaa-ccc-ccc-bbb)
3> (6,bbb-aaa-ccc-ccc-bbb-aaa)

我的程序:

public class WordCountDistinctSocketFilterQEP {
    public WordCountDistinctSocketFilterQEP() throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // @formatter:off
        env.socketTextStream("localhost", 9000)
                .flatMap(new SplitterFlatMap())
                .keyBy(new MyKeySelector())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .reduce(new CountReduceFunction())
                .map(new SwapMapFunction())
                .keyBy(0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // TESTING REMOVING THIS WINDOW
                .reduce(new CountDistinctFunction())
                .print();
        // @formatter:on

        String executionPlan = env.getExecutionPlan();
        System.out.println("ExecutionPlan ........................ ");
        System.out.println(executionPlan);
        System.out.println("........................ ");
        // dataStream.print();

        env.execute("WordCountDistinctSocketFilterQEP");
    }

    public static class SwapMapFunction implements MapFunction<Tuple2<String, Integer>, Tuple2<Integer, String>> {
        private static final long serialVersionUID = 5148172163266330182L;

        @Override
        public Tuple2<Integer, String> map(Tuple2<String, Integer> value) throws Exception {
            return Tuple2.of(1, value.f0);
        }
    }

    public static class SplitterFlatMap implements FlatMapFunction<String, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 3121588720675797629L;

        @Override
        public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception {
            for (String word : sentence.split(" ")) {
                out.collect(new Tuple2<String, Integer>(word, 1));
            }
        }
    }

    public static class MyKeySelector implements KeySelector<Tuple2<String, Integer>, String> {
        private static final long serialVersionUID = 2787589690596587044L;

        @Override
        public String getKey(Tuple2<String, Integer> value) throws Exception {
            return value.f0;
        }
    }

    public static class CountReduceFunction implements ReduceFunction<Tuple2<String, Integer>> {
        private static final long serialVersionUID = 8541031982462158730L;

        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2)
                throws Exception {
            return Tuple2.of(value1.f0, value1.f1 + value2.f1);
        }
    }

    public static class CountDistinctFunction implements ReduceFunction<Tuple2<Integer, String>> {
        private static final long serialVersionUID = -7077952757215699563L;

        @Override
        public Tuple2<Integer, String> reduce(Tuple2<Integer, String> value1, Tuple2<Integer, String> value2)
                throws Exception {
            return Tuple2.of(value1.f0 + value2.f0, value1.f1 + "-" + value2.f1);
        }
    }
}

Answer 1

ReduceFunctions与 Collections 更好地合作( Maps 、 Lists 、 Sets )。如果将每个单词映射到一个元素 Set ，你可以写一个 ReduceFunction运行于 Set<String>然后你可以用一个 ReduceFunction 来做到这一点而不是两个。

所以有splitterFlatMap返回一系列由一个元素组成的长 Set<String> , MyKeySelector返回每个集合的第一个元素。窗口函数很好，更改reduce函数以匹配Set<String>类型，函数的核心是 value1.addAll(value2) 。此时，您已经获得了输入中所有唯一单词的集合，这些单词分布在您正在运行的多个并行任务中。根据完成后您将所有这些数据放在哪里，这可能就足够了。否则，您可以在其末尾放置一个全局窗口，并再次使用相同的reduce函数(解释如下)

你的第二个问题是这不会按原样扩展。在某种程度上，这更像是一个哲学问题。如果不让每个并行实例都与其他实例通信，您就无法真正获得跨并行实例的全局计数。不过，您可以做的是通过实际单词对拆分单词流进行键控，然后使用(并行)键控、窗口 ReduceFunction获取每个键组中不同单词的列表。然后你可以再吃一个ReduceFunction这不是并行的，它结合了并行结果的结果。您还希望第二个窗口也打开； WindowFunctions在触发之前等待所有上游运算符达到正确的水印，因此窗口将确保您的非并行运算符接收来自每个并行运算符的输入。非并行运算符上的聚合是简单的串联，因为一开始的键控保证给定的单词将恰好存在于一个并行槽中。

很明显，单个非并行运算符可能会出现瓶颈，但负载规模与不同单词的总数有关，实际上，由于英语的工作方式，负载规模可能仅限于 10k 单词左右.