欢送拜访我的GitHub

https://github.com/zq2599/blog_demos

内容:所有原创文章分类汇总及配套源码,波及Java、Docker、Kubernetes、DevOPS等;

Flink处理函数实战系列链接

  1. 深刻理解ProcessFunction的状态操作(Flink-1.10);
  2. ProcessFunction;
  3. KeyedProcessFunction类;
  4. ProcessAllWindowFunction(窗口解决);
  5. CoProcessFunction(双流解决);

本篇概览

本文是《Flink处理函数实战》系列的第三篇,上一篇《Flink处理函数实战之二:ProcessFunction类》学习了最简略的ProcessFunction类,明天要理解的KeyedProcessFunction,以及该类带来的一些个性;

对于KeyedProcessFunction

通过比照类图能够确定,KeyedProcessFunction和ProcessFunction并无间接关系:

KeyedProcessFunction用于解决KeyedStream的数据汇合,相比ProcessFunction类,KeyedProcessFunction领有更多个性,官网文档如下图红框,状态解决和定时器性能都是KeyedProcessFunction才有的:

介绍结束,接下来通过实例来学习吧;

版本信息

  1. 开发环境操作系统:MacBook Pro 13寸, macOS Catalina 10.15.3
  2. 开发工具:IDEA ULTIMATE 2018.3
  3. JDK:1.8.0_211
  4. Maven:3.6.0
  5. Flink:1.9.2

源码下载

如果您不想写代码,整个系列的源码可在GitHub下载到,地址和链接信息如下表所示(https://github.com/zq2599/blo...:

名称链接备注
我的项目主页https://github.com/zq2599/blo...该我的项目在GitHub上的主页
git仓库地址(https)https://github.com/zq2599/blo...该我的项目源码的仓库地址,https协定
git仓库地址(ssh)git@github.com:zq2599/blog_demos.git该我的项目源码的仓库地址,ssh协定

这个git我的项目中有多个文件夹,本章的利用在<font color="blue">flinkstudy</font>文件夹下,如下图红框所示:

实战简介

本次实战的指标是学习KeyedProcessFunction,内容如下:

  1. 监听本机9999端口,获取字符串;
  2. 将每个字符串用空格分隔,转成Tuple2实例,f0是分隔后的单词,f1等于1;
  3. 上述Tuple2实例用f0字段分区,失去KeyedStream;
  4. KeyedSteam转入自定义KeyedProcessFunction解决;
  5. 自定义KeyedProcessFunction的作用,是记录每个单词最新一次呈现的工夫,而后建一个十秒的定时器,十秒后如果发现这个单词没有再次出现,就把这个单词和它呈现的总次数发送到上游算子;

编码

  1. 持续应用《Flink处理函数实战之二:ProcessFunction类》一文中创立的工程flinkstudy;
  2. 创立bean类CountWithTimestamp,外面有三个字段,为了方便使用间接设为public:
package com.bolingcavalry.keyedprocessfunction;public class CountWithTimestamp {    public String key;    public long count;    public long lastModified;}
  1. 创立FlatMapFunction的实现类Splitter,作用是将字符串宰割后生成多个Tuple2实例,f0是分隔后的单词,f1等于1:
package com.bolingcavalry;import org.apache.flink.api.common.functions.FlatMapFunction;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.util.Collector;import org.apache.flink.util.StringUtils;public class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> {    @Override    public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {        if(StringUtils.isNullOrWhitespaceOnly(s)) {            System.out.println("invalid line");            return;        }        for(String word : s.split(" ")) {            collector.collect(new Tuple2<String, Integer>(word, 1));        }    }}
  1. 最初是整个逻辑性能的主体:ProcessTime.java,这外面有自定义的KeyedProcessFunction子类,还有程序入口的main办法,代码在上面列出来之后,还会对要害局部做介绍:
package com.bolingcavalry.keyedprocessfunction;import com.bolingcavalry.Splitter;import org.apache.flink.api.common.state.ValueState;import org.apache.flink.api.common.state.ValueStateDescriptor;import org.apache.flink.api.java.tuple.Tuple;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.configuration.Configuration;import org.apache.flink.streaming.api.TimeCharacteristic;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;import org.apache.flink.streaming.api.functions.KeyedProcessFunction;import org.apache.flink.streaming.api.watermark.Watermark;import org.apache.flink.util.Collector;import java.text.SimpleDateFormat;import java.util.Date;/** * @author will * @email zq2599@gmail.com * @date 2020-05-17 13:43 * @description 体验KeyedProcessFunction类(工夫类型是解决工夫) */public class ProcessTime {    /**     * KeyedProcessFunction的子类,作用是将每个单词最新呈现工夫记录到backend,并创立定时器,     * 定时器触发的时候,查看这个单词间隔上次呈现是否曾经达到10秒,如果是,就发射给上游算子     */    static class CountWithTimeoutFunction extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>> {        // 自定义状态        private ValueState<CountWithTimestamp> state;        @Override        public void open(Configuration parameters) throws Exception {            // 初始化状态,name是myState            state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));        }        @Override        public void processElement(                Tuple2<String, Integer> value,                Context ctx,                Collector<Tuple2<String, Long>> out) throws Exception {            // 获得以后是哪个单词            Tuple currentKey = ctx.getCurrentKey();            // 从backend获得以后单词的myState状态            CountWithTimestamp current = state.value();            // 如果myState还从未没有赋值过,就在此初始化            if (current == null) {                current = new CountWithTimestamp();                current.key = value.f0;            }            // 单词数量加一            current.count++;            // 取以后元素的工夫戳,作为该单词最初一次呈现的工夫            current.lastModified = ctx.timestamp();            // 从新保留到backend,包含该单词呈现的次数,以及最初一次呈现的工夫            state.update(current);            // 为以后单词创立定时器,十秒后后触发            long timer = current.lastModified + 10000;            ctx.timerService().registerProcessingTimeTimer(timer);            // 打印所有信息,用于核查数据正确性            System.out.println(String.format("process, %s, %d, lastModified : %d (%s), timer : %d (%s)\n\n",                    currentKey.getField(0),                    current.count,                    current.lastModified,                    time(current.lastModified),                    timer,                    time(timer)));        }        /**         * 定时器触发后执行的办法         * @param timestamp 这个工夫戳代表的是该定时器的触发工夫         * @param ctx         * @param out         * @throws Exception         */        @Override        public void onTimer(                long timestamp,                OnTimerContext ctx,                Collector<Tuple2<String, Long>> out) throws Exception {            // 获得以后单词            Tuple currentKey = ctx.getCurrentKey();            // 获得该单词的myState状态            CountWithTimestamp result = state.value();            // 以后元素是否曾经间断10秒未呈现的标记            boolean isTimeout = false;            // timestamp是定时器触发工夫,如果等于最初一次更新工夫+10秒,就示意这十秒内曾经收到过该单词了,            // 这种间断十秒没有呈现的元素,被发送到上游算子            if (timestamp == result.lastModified + 10000) {                // 发送                out.collect(new Tuple2<String, Long>(result.key, result.count));                isTimeout = true;            }            // 打印数据,用于核查是否合乎预期            System.out.println(String.format("ontimer, %s, %d, lastModified : %d (%s), stamp : %d (%s), isTimeout : %s\n\n",                    currentKey.getField(0),                    result.count,                    result.lastModified,                    time(result.lastModified),                    timestamp,                    time(timestamp),                    String.valueOf(isTimeout)));        }    }    public static void main(String[] args) throws Exception {        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        // 并行度1        env.setParallelism(1);        // 解决工夫        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);        // 监听本地9999端口,读取字符串        DataStream<String> socketDataStream = env.socketTextStream("localhost", 9999);        // 所有输出的单词,如果超过10秒没有再次出现,都能够通过CountWithTimeoutFunction失去        DataStream<Tuple2<String, Long>> timeOutWord = socketDataStream                // 对收到的字符串用空格做宰割,失去多个单词                .flatMap(new Splitter())                // 设置工夫戳分配器,用以后工夫作为工夫戳                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {                    @Override                    public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {                        // 应用以后零碎工夫作为工夫戳                        return System.currentTimeMillis();                    }                    @Override                    public Watermark getCurrentWatermark() {                        // 本例不须要watermark,返回null                        return null;                    }                })                // 将单词作为key分区                .keyBy(0)                // 按单词分区后的数据,交给自定义KeyedProcessFunction解决                .process(new CountWithTimeoutFunction());        // 所有输出的单词,如果超过10秒没有再次出现,就在此打印进去        timeOutWord.print();        env.execute("ProcessFunction demo : KeyedProcessFunction");    }    public static String time(long timeStamp) {        return new SimpleDateFormat("yyyy-MM-dd hh:mm:ss").format(new Date(timeStamp));    }}

上述代码有几处须要重点关注的:

  1. 通过assignTimestampsAndWatermarks设置工夫戳的时候,getCurrentWatermark返回null,因为用不上watermark;
  2. processElement办法中,state.value()能够获得以后单词的状态,state.update(current)能够设置以后单词的状态,这个性能的详情请参考《深刻理解ProcessFunction的状态操作(Flink-1.10)》;
  3. registerProcessingTimeTimer办法设置了定时器的触发工夫,留神这里的定时器是基于processTime,和官网demo中的eventTime是不同的;
  4. 定时器触发后,onTimer办法被执行,外面有这个定时器的全副信息,尤其是入参timestamp,这是本来设置的该定时器的触发工夫;

验证

  1. 在控制台执行命令nc -l 9999,这样就能够从控制台向本机的9999端口发送字符串了;
  2. 在IDEA上间接执行ProcessTime类的main办法,程序运行就开始监听本机的9999端口了;
  3. 在后面的控制台输出aaa,而后回车,期待十秒后,IEDA的控制台输入以下信息,从后果可见合乎预期:

  1. 持续输出aaa再回车,间断两次,两头距离不要超过10秒,后果如下图,可见每一个Tuple2元素都有一个定时器,然而第二次输出的aaa,其定时器在登程前,aaa的最新呈现工夫就被第三次输出的操作给更新了,于是第二次输出aaa的定时器中的比照操作发现此时距aaa的最近一次(即第三次)呈现还未达到10秒,所以第二个元素不会发射到上游算子:

  1. 上游算子收到的所有超时信息会打印进去,如下图红框,只打印了数量等于1和3的记录,等于2的时候因为在10秒内再次输出了aaa,因而没有超时接管,不会在上游打印:


至此,KeyedProcessFunction处理函数的学习就实现了,其状态读写和定时器操作都是很实用能力,心愿本文能够给您提供参考;

你不孤独,欣宸原创一路相伴

  1. Java系列
  2. Spring系列
  3. Docker系列
  4. kubernetes系列
  5. 数据库+中间件系列
  6. DevOps系列

欢送关注公众号:程序员欣宸

微信搜寻「程序员欣宸」,我是欣宸,期待与您一起畅游Java世界...
https://github.com/zq2599/blog_demos