ApacheFlink流媒体窗口字数

Question

我有以下代码来计算socketTextStream中的单词。累积字数和时间窗字数都是必需的。该程序存在累积计数始终与窗口计数相同的问题。为什么会出现这个问题？根据加窗计数计算累积计数的正确方法是什么？

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
final HashMap<String, Integer> cumulateCounts = new HashMap<String, Integer>();

final DataStream<Tuple2<String, Integer>> counts = env
            .socketTextStream("localhost", 9999)
            .flatMap(new Splitter())
            .window(Time.of(5, TimeUnit.SECONDS))
            .groupBy(0).sum(1)
            .flatten();

counts.print();

counts.addSink(new SinkFunction<Tuple2<String, Integer>>() {
    @Override
    public void invoke(Tuple2<String, Integer> value) throws Exception {
        String word = value.f0;
        Integer delta_count = value.f1;
        Integer count = cumulateCounts.get(word);
        if (count == null)
            count = 0;
        count = count + delta_count;
        cumulateCounts.put(word, count);
        System.out.println("(" + word + "," + count.toString() + ")");
    }
});

Answer 1

您应该首先分组，并在键控数据流上应用窗口（您的代码适用于Flink 0.9.1，但Flink 0.10.0中的新API对此要求严格）：

final DataStream<Tuple2<String, Integer>> counts = env
        .socketTextStream("localhost", 9999)
        .flatMap(new Splitter())
        .groupBy(0)
        .window(Time.of(5, TimeUnit.SECONDS)).sum(1)
        .flatten();

如果在未设置关键帧的数据流上应用窗口，则在一台计算机上只有一个单线程窗口操作符（即，没有并行性）在整个数据流上构建窗口（在Flink 0.9.1中，此全局窗口可以通过groupBy（））拆分为子窗口，但在Flink 0.10.0中，这将不再有效）。要计算字数，您需要为每个不同的键值构建一个窗口，即，首先为每个键值获取一个子流（通过groupBy（）），并在每个子流上应用一个窗口操作符（因此，您可以为每个子流拥有一个自己的窗口操作符实例，允许并行执行）。

对于全局（累积）计数，您可以简单地应用group by（）. sum（）构造。首先，流被分成子流（每个键值一个）。其次，计算整个流的总和。因为流没有窗口化，所以计算（累积）中的总和并为每个传入的元组更新（更详细地说，总和的初始结果值为零，并且每个元组的结果更新为结果=tuple.value ）。每次调用求和后，都会发出新的当前结果。

在您的代码中，您不应使用特殊的接收器函数，而应按如下操作：

counts.groupBy(0).sum(1).print();