Flink的aggregate()方法一般是通过实现AggregateFunction接口对数据流进行聚合计算的场景。例如，在使用 Flink 的DataStream API时，用户经常需要对输入数据进行分组操作，并按照一组 key对数据进行汇总、运算或聚合计算。对于这些场景，可以使用 aggregate()方法来实现聚合计算。通过指定一个AggregateFunction类型的函数作为聚合操作来调用aggregate()方法，可以对元素流进行聚合和处理，生成新的输出流。在具体应用中，根据不同的业务需求，可以根据实际情况选择不同类型的AggregateFunction来完成聚合计算任务。
接下来先对AggregateFunction中的需要实现的4个方法进行说明
1. createAccumulator()
此方法用于创建累加器，并将其初始化为默认值
2. add()
此方法将输入的元素添加到累加器，返回更新后的累加器
3. getResult()
此方法用于从累加器中提取操作的结果
4. merge()
此方法将两个累加器合并为一个新的累加器

下面在通过代码实例说明AggregateFunction的使用,这里都以Tuple2类型作为举例说明

• 求平均值

public static class AverageAggregate implements AggregateFunction<Tuple2<String, Double>, Tuple2<String, Double>, Tuple2<String, Double>> {@Overridepublic Tuple2<String, Double> createAccumulator() {// 先将累加器进行初始化,这里给了一个""作为key, 0.0作为值return Tuple2.of("", 0.0); }@Overridepublic Tuple2<String, Double> add(Tuple2<String, Double> value, Tuple2<String, Double> accumulator) {// 这里的实现是将输入的元素和累加器中的元素相加，并返回一个新的元素return Tuple2.of(value.f0, accumulator.f1 + value.f1); }@Overridepublic Tuple2<String, Double> getResult(Tuple2<String, Double> accumulator) {// 这里返回一个包含平均值的 Tuple2 对象，这里是将累加器中的元素除以2，然后返回一个新元素。return Tuple2.of(accumulator.f0, accumulator.f1 / 2.0);}@Overridepublic Tuple2<String, Double> merge(Tuple2<String, Double> a, Tuple2<String, Double> b) {// 这里是将两个累加器中的元素相加并除以2，然后返回一个新的元素对。return Tuple2.of(a.f0, (a.f1 + b.f1) / 2);}}

• 求最大值

public class MaxAggregateFunction implements AggregateFunction<Tuple2<String, Double>, Tuple2<String, Double>, Tuple2<String, Double>> {@Overridepublic Tuple2<String, Double> createAccumulator() {return Tuple2.of("", Double.MIN_VALUE); // 将累加器初始化为最小值}@Overridepublic Tuple2<String, Double> add(Tuple2<String, Double> value, Tuple2<String, Double> accumulator) {if (value.f1 > accumulator.f1) {return Tuple2.of(value.f0, value.f1);} else {return accumulator;}}@Overridepublic Tuple2<String, Double> getResult(Tuple2<String, Double> accumulator) {return accumulator; // 返回最大值}@Overridepublic Tuple2<String, Double> merge(Tuple2<String, Double> a,Tuple2<String, Double> b) {if (a.f1 > b.f1) {return a;} else {return b;}}
}

• 求最小值

public class MinAggregateFunction implements AggregateFunction<Tuple2<String, Double>, Tuple2<String, Double>, Tuple2<String, Double>> {@Overridepublic Tuple2<String, Double> createAccumulator() {return Tuple2.of("", Double.MAX_VALUE); // 将累加器初始化为最大值}@Overridepublic Tuple2<String, Double> add(Tuple2<String, Double> value, Tuple2<String, Double> accumulator) {if (value.f1 < accumulator.f1) {return Tuple2.of(value.f0, value.f1);} else {return accumulator;}}@Overridepublic Tuple2<String, Double> getResult(Tuple2<String, Double> accumulator) {return accumulator; // 返回最小值}@Overridepublic Tuple2<String, Double> merge(Tuple2<String, Double> a,Tuple2<String, Double> b) {if (a.f1 < b.f1) {return a;} else {return b;}}
}

• 求和

public class SumAggregateFunction implements AggregateFunction<Tuple2<String, Double>, Double, Double> {@Overridepublic Double createAccumulator() {return 0.0; // 将累加器初始化为0}@Overridepublic Double add(Tuple2<String, Double> value, Double accumulator) {return value.f1 + accumulator; // 将输入元素和累加器中的元素相加}@Overridepublic Double getResult(Double accumulator) {return accumulator; // 返回总和}@Overridepublic Double merge(Double a,Double b) {return a + b; // 合并两个累加器中的元素相加}
}

以上代码就是通过实现AggregateFunction接口,自定义不同的逻辑达到求平均值、最大值、最小值、总和的目的。

• 方法调用演示

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
/*** @Author: J* @Version: 1.0* @CreateTime: 2023/2/1* @Description: 测试**/
public class Demo1 {public static void main(String[] args) throws Exception {Properties prop = new Properties();StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 这里以kafka作为数据源KafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers("lx01:9092").setTopics("topic-01").setGroupId("g02").setStartingOffsets(OffsetsInitializer.latest()).setValueOnlyDeserializer(new SimpleStringSchema()).build();DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "KafkaSource");// 先将数据转成需要的Tuple2的形式SingleOutputStreamOperator<Tuple2<String, Double>> mapStream = stream.map((MapFunction<String, Tuple2<String, Double>>) value -> {JSONObject data = JSONObject.parseObject(JSONObject.parseObject(value).get("data").toString());return Tuple2.of(data.getString("gender"), data.getDouble("salary"));}).returns(TypeInformation.of(new TypeHint<Tuple2<String, Double>>() {}));// 这里先通过keyBy将数据根据性别进行分组,然后5秒为一个窗口,再求不同性别对应的工资平均值SingleOutputStreamOperator<Tuple2<String, Double>> avg = mapStream.keyBy((KeySelector<Tuple2<String, Double>, String>) value -> {String key = value.f0;return key;}).window(TumblingProcessingTimeWindows.of(Time.seconds(5))).aggregate(new AverageAggregate()); // 这里调用平均值的AggregateFunctionavg.print();env.execute();}

• AggregateFunction的这4个方法在flink中执行的原理
  在 Flink 中AggregateFunction的这四个方法在执行过程中会被转化为对应的内部Function对象，用于Flink的运行时执行计算。

  1. 在数据输入流进入 Flink 的过程中，Flink会为每一个key创建对应的累加器。键值对流会按照键所在的组进行分区，然后把每一个组的所有元素分配到一个task slot中，并为每个key创建一个累加器。累加器的类型是任务的一个状态的函数，Flink根据累加器函数的类型来决定使用哪种累加器。
  2. 当每个数据元素输入到累加器中时，add()方法会被调用。add()方法对输入的元素进行变换，然后更新累加器中的结果，返回新的结果给Flink 的相应算子。
  3. 在结果计算完毕后，getResult()方法将被调用，并将结果返回给 Flink。最后，如果有多个累加器需要合并的情况，Flink 会调用merge()方法将结果进行合并。通过这样的执行机制，AggregateFunction对象可以更加灵活快捷地处理数据。

• 累加器的选择
  当 Flink 创建累加器时，它会根据AggregateFunction的类型来确定使用哪种类型的累加器。具体来说，Flink支持两种类型的累加器：heap-based和incremental。
  heap-based累加器需要在内存中存储完整的所有元素，对于数据量较小的情况，它可以提供最好的性能。对于数据量更大的情况，它可能会导致内存不足的问题。
  incremental累加器可以在输入元素上进行增量操作，并在内存中保存仅仅是必要的元素。它可以处理更大的数据量，并且在内存使用上更加高效。在使用增量式累加器时，用户需要重写accumulate()和retract()方法。
  根据AggregateFunction的类型，Flink会自动选择合适的累加器类型来进行计算，以提高计算的效率和性能。

以上就是对aggregate方法的使用讲解及简单的原理介绍