说明

是 ReduceFunction 的通用版本

应用场景

电商平台，一般都要统计热门商品，或者哪个url的访问最大，典型双11，如果首页某个商品访问量没达标，要立即撤下，不能让需求较小的商品占据首页黄金位置。

需要设定窗口，统计每个url的访问量，然后再去排序

调用方式

stream
 .keyBy( ... )
 .window( ... )
 .aggregate( AggregateFunction 对象);

接口定义

@PublicEvolving
public interface AggregateFunction<IN, ACC, OUT> extends Function, Serializable {
    //创建初始累加器
    ACC createAccumulator();

    //将输入元素，进行增量聚合
    ACC add(IN var1, ACC var2);

    // 返回结果，一个窗口，只会调用一次
    OUT getResult(ACC var1);

    //在 session window 会用到。两个累加器合并为一个累加器
    ACC merge(ACC var1, ACC var2);
}

泛型具有三种类型：

输入类型（IN）
累加器类型（ACC）
输出类型（OUT）

这样设计主要是为了解决：输入类型、累加器类型 和 输出类型 不一致的问题

同时 累加器类型 可以自定义，我们可以在 累加器类型 里构建我们想要的数据结构。
比如我们要计算一个窗口内某个字段的平均值，那么ACC中要保存总和以及个数

案例

统计各书出现的次数

数据

id,文章标题
111,java从入门到入坟
222,mysql从入门到入坟
333,hadoop从入门到放弃
333,hadoop从入门到放弃
444,flink从入门到放弃

思路

根据字符串，进行拆分，并封装到对象中
keyBy() 根据书名进行分组
实现 aggregate()，统计各本书出现的次数

连接kafka

代码

package day4;

import org.apache.flink.api.common.RuntimeExecutionMode;
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.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;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class Std6agg {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        env.setParallelism(1);

        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");

        // 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("guanggao", new SimpleStringSchema(), pro);

        DataStreamSource<String> text = env.addSource(fkc);

        SingleOutputStreamOperator<Result> agg=text.map(new MapFunImpl())
                .keyBy(value -> value.getTitle())//流处理中的分组
                .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
                .aggregate(new AggFunImpl());
        agg.print();
        //启动并等待程序结束，可传入任务名
        env.execute();
    }

    static class MapFunImpl implements MapFunction<String, Article> {
        @Override
        public Article map(String s) throws Exception {
            String[] arr=s.split(",");
            Article article=new Article();
            article.setId(arr[0]);
            article.setTitle(arr[1]);
            return article;
        }
    }

    static class AggFunImpl implements AggregateFunction<Article, Result, Result> {

        @Override
        public Result createAccumulator() {
            return new Result("",0);
        }

        @Override
        public Result add(Article value, Result accumulator) {
            accumulator.setKey(value.getTitle());
            accumulator.setNum( accumulator.getNum()  +  1  );
            return accumulator;
        }

        @Override
        public Result getResult(Result accumulator) {
            return accumulator;
        }

        @Override
        public Result merge(Result a, Result b) {

            return null;
        }
    }

    //文章封装类
    static class Article{
        String id;
        String title;
        public String getId() {
            return id;
        }
        public void setId(String id) {
            this.id = id;
        }
        public String getTitle() {
            return title;
        }
        public void setTitle(String title) {
            this.title = title;
        }

        @Override
        public String toString() {
            return "Article{" +
                    "id='" + id + '\'' +
                    ", title='" + title + '\'' +
                    '}';
        }
    }
    //结果封装类
    static class Result{
        String key;
        Integer num;
        public Result(String key, Integer num) {
            this.key = key;
            this.num = num;
        }
        public String getKey() {
            return key;
        }
        public void setKey(String key) {
            this.key = key;
        }
        public Integer getNum() {
            return num;
        }
        public void setNum(Integer num) {
            this.num = num;
        }
        @Override
        public String toString() {
            return "Result{" +
                    "key='" + key + '\'' +
                    ", num=" + num +
                    '}';
        }
    }
}

启动 kafka 环境

启动 hadoop1、hadoop2、hadoop3 虚拟机

启动zookeeper

/program/bin/zk.sh start

启动kafka

/program/bin/kafka.sh start

创建主题 `guanggao`

flink 从该主题读取数据：

/program/kafka_2.12-2.4.1/bin/kafka-topics.sh --create --zookeeper hadoop1:2181 --replication-factor 3 --partitions 3 --topic guanggao

注意： zookeeper 的主题名称不要有 . 或 _，详见链接

启动 guanggao 主题生产者

启动另一个窗口执行下面命令：

/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092 --topic guanggao

在上面的环境中，每间隔 5s 输入：

111,java从入门到入坟
222,mysql从入门到入坟
333,hadoop从入门到放弃
333,hadoop从入门到放弃
444,flink从入门到放弃

观察 idea 控制台，显示各本书出现的次数

连接 socket

代码

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
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;

public class Std12_agg {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        DataStream<String> dataStream = env.socketTextStream("hadoop1", 9999);
        SingleOutputStreamOperator<Article> soso=dataStream.map(new MapFunction<String, Article>() {
            @Override
            public Article map(String s) throws Exception {
                String[] arr=s.split(",");
                Article article=new Article();
                article.setId(arr[0]);
                article.setTitle(arr[1]);
                return article;
            }
        });
        KeyedStream<Article,String> ks=soso.keyBy(new KeySelector<Article,String>(){
            /**
             *
             * @param t：传入的元祖，格式：(广告id,点击数量)
             * @return
             * @throws Exception
             */
            @Override
            public String getKey(Article t) throws Exception {
                return t.getTitle();
            }
        });//流处理中的分组
        //统计各个key出现的次数
        SingleOutputStreamOperator<Result> agg = ks.window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
                .aggregate(new AggFunImpl());
        agg.print();
        //启动并等待程序结束，可传入任务名
        env.execute();
    }

    static class AggFunImpl implements AggregateFunction<Article, Result, Result> {

        @Override
        public Result createAccumulator() {
            return new Result("",0);
        }

        @Override
        public Result add(Article value, Result accumulator) {
            accumulator.setKey(value.getTitle());
            accumulator.setNum( accumulator.getNum()  +  1  );
            return accumulator;
        }

        @Override
        public Result getResult(Result accumulator) {
            return accumulator;
        }

        @Override
        public Result merge(Result a, Result b) {
            a.setNum(  a.getNum() + b.getNum() );
            return a;
        }
    }
    //文章封装类
    static class Article{
        String id;
        String title;
        public String getId() {
            return id;
        }
        public void setId(String id) {
            this.id = id;
        }
        public String getTitle() {
            return title;
        }
        public void setTitle(String title) {
            this.title = title;
        }
    }
    //结果封装类
    static class Result{
        String key;
        Integer num;
        public Result(String key, Integer num) {
            this.key = key;
            this.num = num;
        }
        public String getKey() {
            return key;
        }
        public void setKey(String key) {
            this.key = key;
        }
        public Integer getNum() {
            return num;
        }
        public void setNum(Integer num) {
            this.num = num;
        }
        @Override
        public String toString() {
            return "Result{" +
                    "key='" + key + '\'' +
                    ", num=" + num +
                    '}';
        }
    }
}

测试

在linux 执行下面命令：

nc -lk 9999

启动java程序

在 nc 窗口输入下面内容：

111,java从入门到入坟
222,mysql从入门到入坟
333,hadoop从入门到放弃
333,hadoop从入门到放弃
444,flink从入门到放弃