Flink 中有哪些类型的窗口函数？

推荐答案

在 Apache Flink 中，窗口函数主要分为以下几类：

1. ReduceFunction：对窗口中的元素进行聚合操作，输出一个单一的结果。
2. AggregateFunction：与 ReduceFunction 类似，但提供了更灵活的聚合方式，支持中间状态和最终结果的输出。
3. ProcessWindowFunction：提供了对窗口内所有元素的完全访问权限，允许自定义复杂的窗口计算逻辑。
4. WindowFunction：与 ProcessWindowFunction 类似，但功能较为简单，通常用于较早期的 Flink 版本。

本题详细解读

ReduceFunction

ReduceFunction 是一种简单的窗口函数，它通过对窗口中的元素进行两两聚合，最终输出一个单一的结果。例如，计算窗口内所有元素的和：

// www.javascriptcn.com code example
DataStream<Tuple2<String, Integer>> input = ...;
input
    .keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) {
            return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
        }
    });

AggregateFunction

AggregateFunction 提供了更灵活的聚合方式，允许在聚合过程中维护中间状态，并最终输出一个结果。例如，计算窗口内元素的平均值：

// www.javascriptcn.com code example
DataStream<Tuple2<String, Integer>> input = ...;
input
    .keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .aggregate(new AggregateFunction<Tuple2<String, Integer>, Tuple2<Integer, Integer>, Double>() {
        @Override
        public Tuple2<Integer, Integer> createAccumulator() {
            return new Tuple2<>(0, 0);
        }

        @Override
        public Tuple2<Integer, Integer> add(Tuple2<String, Integer> value, Tuple2<Integer, Integer> accumulator) {
            return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1);
        }

        @Override
        public Double getResult(Tuple2<Integer, Integer> accumulator) {
            return ((double) accumulator.f0) / accumulator.f1;
        }

        @Override
        public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {
            return new Tuple2<>(a.f0 + b.f0, a.f1 + b.f1);
        }
    });

ProcessWindowFunction

ProcessWindowFunction 提供了对窗口内所有元素的完全访问权限，允许自定义复杂的窗口计算逻辑。例如，计算窗口内元素的最大值并输出相关信息：

// www.javascriptcn.com code example
DataStream<Tuple2<String, Integer>> input = ...;
input
    .keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .process(new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {
        @Override
        public void process(String key, Context context, Iterable<Tuple2<String, Integer>> elements, Collector<String> out) {
            int max = Integer.MIN_VALUE;
            for (Tuple2<String, Integer> element : elements) {
                max = Math.max(max, element.f1);
            }
            out.collect("Window: " + context.window() + " max value: " + max);
        }
    });

WindowFunction

WindowFunction 是 ProcessWindowFunction 的前身，功能较为简单，通常用于较早期的 Flink 版本。例如，计算窗口内元素的和并输出相关信息：

// www.javascriptcn.com code example
DataStream<Tuple2<String, Integer>> input = ...;
input
    .keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .apply(new WindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {
        @Override
        public void apply(String key, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) {
            int sum = 0;
            for (Tuple2<String, Integer> in : input) {
                sum += in.f1;
            }
            out.collect("Window: " + window + " sum: " + sum);
        }
    });