案例

通过各个广告的点击数量

广告id	某时刻点击数量
001	5
002	7
003	3
002	6
003	4

需求

每 5 秒统计一次，统计最近 5 秒内，各个广告点击的数量

即：size=5，slide=5

提示：基于时间的滚动窗口

kafka方式实现

代码

package day4;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
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.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class Std3TumblingProcessingTimeWindows {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        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<Tuple2<String, Integer>> soso = text
                .flatMap(new FlatMapFunImpl())
                .keyBy( value -> value.f0) // 根据tuple的第一个元素（广告id）进行分组
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // 每5秒中处理一次，窗口大小是5秒，即：滚动窗口
                .sum(1); // 对 tuple 的第二个元素（索引是1） 求和

        soso.print();

        System.out.println(env.getExecutionPlan());
        env.execute("myjob");
    }

    // kafka 发送的数据格式是 json，我们接收的是普通字符串，需要转换成 json 对象
    static class FlatMapFunImpl implements FlatMapFunction<String,Tuple2<String,Integer>>{
        @Override
        public void flatMap(String value, Collector<Tuple2<String,Integer>> out) throws Exception {
            // 数据样例：001,3
            String[] arr = value.split(",");
            int num = Integer.parseInt(arr[1]);
            // 封装数据为 Tuple2，第一个元素是 广告id，第二个元素是该广告的点击数量
            out.collect(Tuple2.of(arr[0],num));
        }
    }
}

启动 kafka 环境

启动 hadoop1、hadoop2、hadoop3

启动zookeeper

/program/bin/zk.sh start

启动kafka

/program/bin/kafka.sh start

准备 kafka 主题

创建主题 `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 输入：

001,3
002,3
003,3

socket方式实现

通过 nc 发送数据，flink 使用 socket 方式接收数据

代码

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
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 Window统计广告 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//自动模式，根据数据源自行判断
        DataStream<String> dataStream = env.socketTextStream("hadoop1", 9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> soso=dataStream.map(new MapImpl());
        KeyedStream<Tuple2<String, Integer>,String> ks=soso.keyBy(new keyByImpl());//流处理中的分组

        SingleOutputStreamOperator<Tuple2<String, Integer>> soso2=ks.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
            .sum(1);//按照第二个元素聚合

        soso2.print();
        //启动并等待程序结束，可传入任务名
        env.execute("Window WordCount");
    }

    /**
     * 根据 “,” 拆分，返回元祖，格式： (广告id,点击数量)
     * MapFunction<String, Tuple2<String, Integer>> 泛型解释：
     *      String：传入的是字符串
     *      Tuple2<String, Integer>，返回类型。格式：(广告id,点击数量)
     */
    public static class MapImpl implements MapFunction<String, Tuple2<String, Integer>> {

        @Override
        public Tuple2<String, Integer> map(String s) throws Exception {
            String[] arr=s.split(",");
            return Tuple2.of(arr[0],Integer.parseInt(arr[1]));
        }
    }

    /**
     根据第一个元素（即：广告）进行分组
     KeySelector<Tuple2<String, Integer>,String> 泛型解释：
        Tuple2<String, Integer>：传入的是 Tuple类型，也就是上面封装的Tuple，格式：(广告id,点击数量)
        String：传出类型，也就是Tuple中的单词，根据该单词进行分组
     */
    public static class keyByImpl implements KeySelector<Tuple2<String, Integer>,String>{
        /**
         *
         * @param t：传入的元祖，格式：(广告id,点击数量)
         * @return
         * @throws Exception
         */
        @Override
        public String getKey(Tuple2<String, Integer> t) throws Exception {
            return t.f0;
        }
    }
}

测试

启动 nc

需要先启动 nc，因为启动 java代码后，立刻连接 socket 服务器

先执行下面命令：

nc -lk 9999

启动 java 代码

输入

在 nc 窗口中每间隔 5s 输入：

001,3
002,3
003,3

在 idea 控制台中，显示结果：

8> (003,3)
1> (001,3)
7> (002,3)
1> (001,3)
8> (003,3)
7> (002,3)

因为每 5 秒统计一次，统计最近 5 秒内，各个广告点击的数量，所以统计结果一直都是 3

注意： 统计结果中，不会是0，因为 没有数据不会触发

原文出处：/show_1IX20RqPYlKa.html

案例

需求