温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

Flink Aggregate怎么用

发布时间:2021-12-31 10:21:24 来源:亿速云 阅读:475 作者:iii 栏目:大数据

本篇内容主要讲解“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<Tuple3<String, String, Integer>> tuple3List = DataSource.getTuple3ToList();
        DataStream<MyAverageAccumulator> dataStream = env.fromCollection(tuple3List)
                .keyBy((KeySelector<Tuple3<String, String, Integer>, String>) k -> k.f1)
                //按数量窗口滚动,每3个输入窗口数据流,计算一次
                .countWindow(3)
                //只能基于Windowed窗口Stream进行调用
                .aggregate(new AggregateFunction<Tuple3<String, String, Integer>, MyAverageAccumulator, MyAverageAccumulator>() {
                    /**
                     * 创建新累加器,开始聚合计算
                     * @return
                     */
                    @Override
                    public MyAverageAccumulator createAccumulator() {
                        return new MyAverageAccumulator();
                    }

                    /**
                     * 将窗口输入的数据流值添加到窗口累加器,并返回新的累加器值
                     * @param tuple3
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public MyAverageAccumulator add(Tuple3<String, String, Integer> 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怎么用”有了更深的了解,不妨来实际操作一番吧!这里是亿速云网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!

向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI