前言
今天我们主要聊聊flink中的一个接口org.apache.flink.api.common.functions.AggregateFunction，这个类可以接在window流之后，做窗口内的统计计算。
注意：除了这个接口AggregateFunction，flink中还有一个抽象类AggregateFunction：org.apache.flink.table.functions.AggregateFunction，大家不要把这个弄混淆了，接口AggregateFunction我们可以理解为flink中的一个算子，和MapFunction、FlatMapFunction等是同级别的，而抽象类AggregateFunction是用于用户自定义聚合函数的，和max、min之类的函数是同级的。
原理解析
比如我们想实现一个类似sql的功能:
select TUMBLE_START(proctime,INTERVAL '2' SECOND) as starttime,user,count(*) from logs group by user,TUMBLE(proctime,INTERVAL '2' SECOND)
这个sql就是来统计一下每两秒钟的滑动窗口内每个人出现的次数，今天我们就以这个简单的sql的功能为例讲解一下flink的aggregate算子，其实就是我们用程序来实现这个sql的功能。
首先看一下聚合函数的接口:
@PublicEvolving
public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {
ACC createAccumulator();
ACC add(IN value, ACC accumulator);
ACC merge(ACC a, ACC b);
OUT getResult(ACC accumulator);
}
这个接口AggregateFunction里面有4个方法，我们分别来讲解一下。
- AggregateFunction这个类是一个泛型类，这里面有三个参数，IN, ACC, OUT。IN就是聚合函数的输入类型，ACC是存储中间结果的类型，OUT是聚合函数的输出类型。
- createAccumulator
这个方法首先要创建一个累加器，要进行一些初始化的工作，比如我们要进行count计数操作，就要给累加器一个初始值。 - add
add方法就是我们要做聚合的时候的核心逻辑，比如我们做count累加，其实就是来一个数，然后就加一。
类似上面的sql的逻辑，我们在写业务逻辑的时候，可以这么想，进入这方法数的数据都是属于某一个用户的，系统在调用这个方法之前会先进行hash分组，然后不同的用户会重复调用这个方法。所以这个函数的入参是IN类型，返回值是ACC类型 - merge
因为flink是一个分布式计算框架，可能计算是分布在很多节点上同时进行的，比如上述的add操作，可能同一个用户在不同的节点上分别调用了add方法在本地节点对本地的数据进行了聚合操作，但是我们要的是整个结果，整个时候，我们就需要把每个用户各个节点上的聚合结果merge一下，整个merge方法就是做这个工作的，所以它的入参和出参的类型都是中间结果类型ACC。 - getResult
这个方法就是将每个用户最后聚合的结果经过处理之后，按照OUT的类型返回，返回的结果也就是聚合函数的输出结果了。
实例讲解
自定义source
首先我们自定义source生成用户的信息
public static class MySource implements SourceFunction<Tuple2<String,Long>>{
private volatile boolean isRunning = true;
String userids[] = {
"4760858d-2bec-483c-a535-291de04b2247", "67088699-d4f4-43f2-913c-481bff8a2dc5",
"72f7b6a8-e1a9-49b4-9a0b-770c41e01bfb", "dfa27cb6-bd94-4bc0-a90b-f7beeb9faa8b",
"aabbaa50-72f4-495c-b3a1-70383ee9d6a4", "3218bbb9-5874-4d37-a82d-3e35e52d1702",
"3ebfb9602ac07779||3ebfe9612a007979", "aec20d52-c2eb-4436-b121-c29ad4097f6c",
"e7e896cd939685d7||e7e8e6c1930689d7", "a4b1e1db-55ef-4d9d-b9d2-18393c5f59ee"
};
@Override
public void run(SourceContext<Tuple2<String,Long>> ctx) throws Exception{
while (isRunning){
Thread.sleep(10);
String userid = userids[(int) (Math.random() * (userids.length - 1))];
ctx.collect(Tuple2.of(userid, System.currentTimeMillis()));
}
}
@Override
public void cancel(){
isRunning = false;
}
}
自定义聚合函数
public static class CountAggregate
implements AggregateFunction<Tuple2<String,Long>,Integer,Integer>{
@Override
public Integer createAccumulator(){
return 0;
}
@Override
public Integer add(Tuple2<String,Long> value, Integer accumulator){
return ++accumulator;
}
@Override
public Integer getResult(Integer accumulator){
return accumulator;
}
@Override
public Integer merge(Integer a, Integer b){
return a + b;
}
}
自定义结果输出函数
/**
* 这个是为了将聚合结果输出
*/
public static class WindowResult
implements WindowFunction<Integer,Tuple3<String,Date,Integer>,Tuple,TimeWindow>{
@Override
public void apply(
Tuple key,
TimeWindow window,
Iterable<Integer> input,
Collector<Tuple3<String,Date,Integer>> out) throws Exception{
String k = ((Tuple1<String>) key).f0;
long windowStart = window.getStart();
int result = input.iterator().next();
out.collect(Tuple3.of(k, new Date(windowStart), result));
}
}
主流程
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<Tuple2<String,Long>> dataStream = env.addSource(new MySource());
dataStream.keyBy(0).window(TumblingProcessingTimeWindows.of(Time.seconds(2)))
.aggregate(new CountAggregate(), new WindowResult()
).print();
env.execute();
欢迎关注我的公众号
在这里插入图片描述
网友评论