13518219792
本篇内容主要讲解“Flink Aggregate怎么用”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Flink Aggregate怎么用”吧!
站在用户的角度思考问题，与客户深入沟通，找到澄城网站设计与澄城网站推广的解决方案，凭借多年的经验，让设计与互联网技术结合，创造个性化、用户体验好的作品，建站类型包括：成都做网站、成都网站制作、企业官网、英文网站、手机端网站、网站推广、域名与空间、虚拟空间、企业邮箱。业务覆盖澄城地区。
Aggregate算子：提供基于事件窗口进行增量计算的函数。（对输入窗口每个数据流元素递增聚合计算，并将窗口状态与窗口内元素保持在累加器中）
示例环境
java.version: 1.8.x flink.version: 1.11.1
Aggregate.java
import com.flink.examples.DataSource;
import org.apache.flink.api.common.accumulators.AverageAccumulator;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.List;
/**
* @Description Aggregate算子：提供基于事件窗口进行增量计算的函数。（对输入窗口每个数据流元素递增聚合计算，并将窗口状态与窗口内元素保持在累加器中）
*/
public class Aggregate {
/**
* 遍历集合，分别打印不同性别的总人数与平均值
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//Tuple3<姓名，性别（man男，girl女），年龄>
List> tuple3List = DataSource.getTuple3ToList();
DataStream dataStream = env.fromCollection(tuple3List)
.keyBy((KeySelector, String>) k -> k.f1)
//按数量窗口滚动，每3个输入窗口数据流，计算一次
.countWindow(3)
//只能基于Windowed窗口Stream进行调用
.aggregate(new AggregateFunction, MyAverageAccumulator, MyAverageAccumulator>() {
/**
* 创建新累加器，开始聚合计算
* @return
*/
@Override
public MyAverageAccumulator createAccumulator() {
return new MyAverageAccumulator();
}
/**
* 将窗口输入的数据流值添加到窗口累加器，并返回新的累加器值
* @param tuple3
* @param accumulator
* @return
*/
@Override
public MyAverageAccumulator add(Tuple3 tuple3, MyAverageAccumulator accumulator) {
System.out.println("tuple3:" + tuple3.toString());
accumulator.setGender(tuple3.f1);
//此accumulator保含个数统计和值累计两个属性，add方法内会计算窗口内总数与求和
accumulator.add(tuple3.f2);
return accumulator;
}
/**
* 获取累加器聚合结果
* @param accumulator
* @return
*/
@Override
public MyAverageAccumulator getResult(MyAverageAccumulator accumulator) {
return accumulator;
}
/**
* 合并两个累加器，返回合并后的累加器的状态
* @param a
* @param b
* @return
*/
@Override
public MyAverageAccumulator merge(MyAverageAccumulator a, MyAverageAccumulator b) {
a.merge(b);
return a;
}
});
dataStream.print();
env.execute("flink Filter job");
}
/**
* 添加性别属性（此类用于显示不同性别的平均值）
*/
public static class MyAverageAccumulator extends AverageAccumulator{
private String gender;
public String getGender() {
return gender;
}
public void setGender(String gender) {
this.gender = gender;
}
@Override
public String toString() {
//继承父类的this.getLocalValue()方法用于计算并返回平均值
return super.toString() + ", gender to " + gender;
}
}
} 打印结果
tuple3:(张三,man,20) tuple3:(李四,girl,24) tuple3:(刘六,girl,32) tuple3:(王五,man,29) tuple3:(伍七,girl,18) tuple3:(吴八,man,30) 4> AverageAccumulator 24.666666666666668 for 3 elements, gender to girl 2> AverageAccumulator 26.333333333333332 for 3 elements, gender to man
到此，相信大家对“Flink Aggregate怎么用”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
