窗口自定义聚合
Flink中的.window().aggregate()方法是Apache Flink为了处理无界或有界数据流上的窗口聚合操作而提供的一个关键功能。它允许用户在特定时间窗口内对数据流进行分组和聚合计算，该方法结合了窗口分配器（WindowAssigner）来定义数据如何分配到不同的窗口中，以及聚合函数（AggregateFunction）来定义在每个窗口内如何对数据进行聚合。
.window().aggregate()方法详解
窗口分配器（WindowAssigner）：首先，你需要选择或定义一个窗口分配器，它负责将输入的数据流分割成一系列不重叠或重叠的窗口。常见的窗口类型有滚动窗口（TumblingWindows）、滑动窗口（SlidingWindows）、会话窗口（SessionWindows）等。
聚合函数（AggregateFunction）：接着，你需要提供一个聚合函数，这个函数定义了如何将窗口内的元素聚合起来生成一个结果。
AggregateFunction需要实现四个方法：创建累加器的初始值、合并两个累加器、将一个输入元素应用到累加器上，以及从累加器中提取最终结果。使用方式：在Flink的数据流处理中，通常通过如下方式调用
.window().aggregate()方法：首先，通过keyBy()对数据进行分组；然后，选择合适的窗口分配器应用.window()；最后，调用.aggregate(AggregateFunction)来指定聚合逻辑。相对reduce()的优势
虽然**.reduce()方法也能实现某种程度上的聚合，但它在功能和灵活性上与.window().aggregate()**相比有一些局限：
窗口支持：**.reduce()**直接作用于DataStream，不直接涉及窗口概念，因此对于需要按时间窗口进行聚合的场景，.window().aggregate()提供了更自然的支持。它可以轻松地处理时间窗口内的聚合，而.reduce()则需要额外的逻辑来模拟窗口行为。
灵活性和复杂度：.aggregate()通过AggregateFunction提供了更多的灵活性，可以定义复杂的聚合逻辑，包括初始化累加器、多次累积和最终结果的转换。相比之下，.reduce()仅能进行简单的累计操作，每次调用时都基于前两个元素的累积结果和下一个元素进行计算，不适合复杂聚合需求。
状态管理：AggregateFunction内部的累加器机制为状态管理提供了更好的抽象，使得处理大规模数据流时的状态管理更加高效和易于理解。这对于维护大规模分布式系统中的状态特别重要。
import cn.guangjun.flink.pojo.Event;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.AggregateFunction;
import java.util.HashMap;
import java.util.LinkedHashMap;
/**
* 自定义聚合函数类，用于处理Event类型的事件数据，将其聚合并返回一个字符串形式的JSON表示
* 实现了AggregateFunction接口，指定Event为输入类型，HashMap<String,Integer>为累加器类型，String为结果类型
*/
public class CustomAggregate implements AggregateFunction<Event, HashMap<String,Integer>, String> {
/**
* 创建累加器
* 返回一个空的LinkedHashMap作为累加器，用于收集和保存中间结果
* @return 一个空的LinkedHashMap作为初始累加器
*/
@Override
public HashMap<String,Integer> createAccumulator() {
return new LinkedHashMap<>();
}
/**
* 将事件数据添加到累加器中
* 如果累加器中不存在该事件名称，则添加新的条目；如果已存在，则更新其时间戳总和
* @param event 待添加的事件数据
* @param accumulator 当前的累加器，用于收集事件数据
* @return 更新后的累加器
*/
@Override
public HashMap<String,Integer> add(Event event, HashMap<String,Integer> accumulator) {
if(!accumulator.containsKey(event.getName())){
accumulator.put(event.getName(), event.getTimeStamp());
} else {
Integer sum = accumulator.get(event.getName());
accumulator.put(event.getName(), sum + event.getTimeStamp());
}
return accumulator;
}
/**
* 从累加器中获取最终结果
* 将累加器转换为JSON字符串格式，作为最终结果返回
* @param accumulator 包含聚合结果的累加器
* @return 聚合结果的JSON字符串表示
*/
@Override
public String getResult(HashMap<String,Integer> accumulator) {
return JSON.toJSONString(accumulator);
}
/**
* 合并两个累加器
* 将两个累加器合并为一个，用于支持并行处理时的中间结果合并
* @param a 第一个累加器
* @param b 第二个累加器
* @return 合并后的累加器
*/
@Override
public HashMap<String,Integer> merge(HashMap<String,Integer> a, HashMap<String,Integer> b) {
HashMap<String,Integer> hashmap = new HashMap<>();
hashmap.putAll(b);
hashmap.putAll(b);
System.out.println("merge:" + JSON.toJSONString(hashmap));
return hashmap;
}
}
import cn.guangjun.flink.pojo.Event;
import cn.guangjun.flink.stream.CustomSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
/**
* AggregateByWinDemo 类用于演示如何在 Flink 流处理中按窗口进行聚合操作
* 该示例展示了如何从自定义数据源读取事件，并根据时间戳和水位线进行窗口聚合操作
*/
public class AggregateByWinDemo {
/**
* 主函数，用于设置 Flink 流处理环境并执行聚合操作
* @param args 命令行参数，本例中未使用
* @throws Exception 如果处理过程中发生错误，将抛出异常
*/
public static void main(String[] args) throws Exception {
// 获取 Flink 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置并行度为 1，以简化示例
env.setParallelism(1);
// 从自定义数据源添加数据流，并为数据流分配时间戳和水位线
SingleOutputStreamOperator<Event> stream = env.addSource(new CustomSource())
.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
/**
* 从事件中提取时间戳
* @param element 当前事件对象
* @param recordTimestamp 当前记录的时间戳，未使用
* @return 从事件中提取的时间戳
*/
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.getTimeStamp();
}
}));
// 对数据流按键进行分组，并定义聚合窗口和聚合函数
stream.keyBy(data -> true)
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.aggregate(new CustomAggregate())
.print();
// 执行流处理任务
env.execute();
}
}