flink聚合函数使用例子

在 Apache Flink 中，聚合函数是用于将流数据进行聚合操作的函数，常用于实现基于时间窗口或者滑动窗口的聚合操作。

下面是一个使用 Flink 聚合函数的例子：

DataStream<Tuple2<String, Integer>> input = ...;

// 按照 key 分组，计算每个 key 出现的次数
DataStream<Tuple2<String, Integer>> result = input
    .keyBy(0)
    .timeWindow(Time.seconds(10))
    .aggregate(new CountAggregator());

// 自定义计数器聚合函数
public static class CountAggregator implements AggregateFunction<Tuple2<String, Integer>, Integer, Integer> {
    @Override
    public Integer createAccumulator() {
        return 0;
    }

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

    @Override
    public Integer getResult(Integer accumulator) {
        return accumulator;
    }

    @Override
    public Integer merge(Integer a, Integer b) {
        return a + b;
    }
}

上述例子中，我们首先定义了一个输入流 input，其中包含了一个 Tuple2 类型的元素，第一个元素为 String 类型的 key，第二个元素为 Integer 类型的 value。

接着我们使用了 keyBy() 方法按照 key 进行分组，并使用 timeWindow() 方法定义了一个大小为 10 秒的时间窗口。

最后我们使用了 aggregate() 方法来对每个窗口中的元素进行聚合操作，其中我们传入了一个自定义的计数器聚合函数 CountAggregator。

CountAggregator 实现了 AggregateFunction 接口，其中 createAccumulator() 方法用于创建一个初始的计数器，add() 方法用于将输入元素累加到计数器中，getResult() 方法用于返回计数器的结果，merge() 方法用于合并两个计数器的结果。

综上所述，上述例子中我们使用了 Flink 自带的 keyBy()、timeWindow() 和 aggregate() 方法，以及自定义的计数器聚合函数 CountAggregator，来实现了按照 key 进行分组并计算每个 key 出现次数的聚合操作。