关于云计算:Flink处理函数实战之三KeyedProcessFunction类

欢送拜访我的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…：

这个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