Flink AggregatingState 实战案例：计算数据流平均值

本文将通过一个简单易懂的代码示例，教你如何使用 Flink 的 AggregatingState 来计算数据流中每个键的平均值。

代码示例javaimport org.apache.flink.api.common.functions.AggregateFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class AggregatingStateExample {

public static void main(String[] args) throws Exception {        // 创建执行环境        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    // 输入数据流        DataStream<Tuple2<String, Integer>> input = env.fromElements(                Tuple2.of('key1', 1),                Tuple2.of('key2', 2),                Tuple2.of('key1', 3),                Tuple2.of('key2', 4),                Tuple2.of('key1', 5)        );

    // 使用 AggregatingState 计算每个 key 的平均值        DataStream<Tuple2<String, Double>> average = input                .keyBy(0) // 按照第一个元素分组                .aggregate(new AverageAggregate());

    // 输出结果        average.print();

    // 执行任务        env.execute('AggregatingState Example');    }

// 自定义 AggregateFunction 计算平均值    public static class AverageAggregate implements AggregateFunction<Tuple2<String, Integer>, Tuple2<Integer, Integer>, Tuple2<String, Double>> {

    @Override        public Tuple2<Integer, Integer> createAccumulator() {            // 初始化累加器，第一个元素存储总和，第二个元素存储计数            return Tuple2.of(0, 0);        }

    @Override        public Tuple2<Integer, Integer> add(Tuple2<String, Integer> value, Tuple2<Integer, Integer> accumulator) {            // 更新累加器            return Tuple2.of(accumulator.f0 + value.f1, accumulator.f1 + 1);        }

    @Override        public Tuple2<String, Double> getResult(Tuple2<Integer, Integer> accumulator) {            // 计算平均值并返回结果            double average = (double) accumulator.f0 / accumulator.f1;            return Tuple2.of('average', average);        }

    @Override        public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {            // 合并两个累加器            return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);        }    }}

代码解读

创建执行环境: StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();2. 创建输入数据流: 使用 env.fromElements() 方法创建一个包含多个 Tuple2<String, Integer> 元素的数据流。3. 按 key 分组: 使用 keyBy(0) 方法按照 Tuple 中的第一个元素（即 key）进行分组。4. 定义 AggregateFunction: 创建一个名为 AverageAggregate 的类，实现 AggregateFunction 接口，用于定义如何计算平均值。 - createAccumulator(): 创建一个累加器，用于存储计算平均值所需的中间结果。 - add(): 定义如何将新数据添加到累加器中。 - getResult(): 定义如何从累加器中获取最终结果。 - merge(): 定义如何合并两个累加器。5. 应用 AggregateFunction: 使用 aggregate(new AverageAggregate()) 方法将自定义的 AverageAggregate 应用到分组后的数据流上。6. 输出结果: 使用 print() 方法将计算结果输出到控制台。7. 执行任务: 使用 env.execute() 方法启动 Flink 任务。

总结

本文提供了一个使用 Flink AggregatingState 计算数据流平均值的简单示例，并对代码进行了详细的解读。希望通过本文的学习，你能够掌握 AggregatingState 的基本使用方法。