关于云计算: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