Apache Flink 减少了许多值而不是一个值答案

【问题标题】：Apache Flink reduce results in many values instead of oneApache Flink 减少了许多值而不是一个值
【发布时间】：2019-05-06 12:55:08
【问题描述】：

我正在尝试在 WindowedStream 上实现 reduce，如下所示：

                .keyBy(t -> t.key)
            .timeWindow(Time.of(15, MINUTES), Time.of(1, MINUTES))
            .reduce(new ReduceFunction<TwitterSentiments>() {
                @Override
                public TwitterSentiments reduce(TwitterSentiments t2, TwitterSentiments t1) throws Exception {
                    t2.positive += t1.positive;
                    t2.neutral += t1.neutral;
                    t2.negative += t1.negative;

                    return t2;
                }
            });

我遇到的问题是，当我调用 stream.print() 时，我得到了许多值（看起来像每个 TwitterSentiments 对象一个，而不是单个聚合对象。

我也尝试过使用这样的 AggregationFunction，但遇到同样的问题：

                .aggregate(new AggregateFunction<TwitterSentiments, Tuple3<Long, Long, Long>, Tuple3<Long, Long, Long>>() {
                @Override
                public Tuple3<Long, Long, Long> createAccumulator() {
                    return new Tuple3<Long, Long, Long>(0L,0L,0L);
                }

                @Override
                public Tuple3<Long, Long, Long> add(TwitterSentiments ts, Tuple3<Long, Long, Long> accumulator) {
                    return new Tuple3<Long, Long, Long>(
                            accumulator.f0 + ts.positive.longValue(),
                            accumulator.f1 + ts.neutral.longValue(),
                            accumulator.f2 + ts.negative.longValue()
                    );
                }

                @Override
                public Tuple3<Long, Long, Long> getResult(Tuple3<Long, Long, Long> accumulator) {
                    return accumulator;
                }

                @Override
                public Tuple3<Long, Long, Long> merge(Tuple3<Long, Long, Long> accumulator1, Tuple3<Long, Long, Long> accumulator2) {
                    return new Tuple3<Long, Long, Long>(
                            accumulator1.f0 + accumulator2.f0,
                            accumulator1.f1 + accumulator2.f1,
                            accumulator1.f2 + accumulator2.f1);
                }
            });

stream.print() 在这些聚合之后仍然会输出很多记录的原因是什么？

【问题讨论】：

您可能正在使用EventTime。您可以检查您的流媒体环境的timeCharacteristic 设置（由env.setStreamTimeCharacteristic 设置）吗？如果您使用EventTime，则时间窗口由事件时间而不是本地机器时间触发。
@David 嗯，感谢您的回答，但似乎这不是问题所在。 timeCharacteristic 默认设置为 ProcessingTime。我尝试使用IngestionTime，但仍然遇到同样的问题。这里还能发生什么？
啊，你能检查一下 print 的输出键吗？它们都不同吗？对于同一个键，在打印输出（翻滚窗口）中必须有超过 1 分钟的间隔。
您应该每分钟每个键获得一个结果。如果您的结果包括密钥和时间，则更容易理解正在发生的事情——您可以通过 ProcessWindowFunction 传递预先聚合的结果来实现这一点。有关示例，请参见 github.com/dataArtisans/flink-training-exercises/blob/master/…。
@DavidAnderson 啊，“每个键”。这就说得通了。你的回答帮助我解决了我的问题。谢谢。

标签： java bigdata apache-flink flink-streaming

【解决方案1】：

如果您不需要每个键的结果，则可以使用 timeWindowAll 生成单个结果。但是，timeWindowAll 不会并行运行。如果您想以更具可扩展性的方式计算结果，您可以这样做：

    .keyBy(t -> t.key)
    .timeWindow(<time specification>)
    .reduce(<reduce function>)
    .timeWindowAll(<same time specification>)
    .reduce(<same reduce function>)

您可能希望 Flink 的运行时足够智能，可以为您执行此并行预聚合（前提是您使用的是 ReduceFunction 或 AggregateFunction），但事实并非如此。

【讨论】：

这正是我最初想做的，谢谢。我创建了一个键控流，因为我想要并行处理，但没有看到键控流也按键聚合。这现在更有意义了。你说得对，我曾期望 flink 为我做这个处理。
嗨，我用timeWindow -> process -> timeWindowAll -> reduce做了一个例子，我想知道我是否可以在Flink中实现我自己的操作符（使用SingleOutputStreamOperator）来帮我做

【解决方案2】：

看来我误解了使用密钥的原因。在我的例子中，我不需要KeyedStream，因为我只希望每分钟只有一个输出，它包括所有记录减少到一个值。我最终在SingleOutputStreamOperator 上使用了.timeWindowAll，并且现在运行我的reduce 可以按预期工作。

【讨论】：