乐趣区

关于云计算:CoProcessFunction实战三部曲之三定时器和侧输出

欢送拜访我的 GitHub

这里分类和汇总了欣宸的全副原创(含配套源码):https://github.com/zq2599/blog_demos

本篇概览

  • 本文是《CoProcessFunction 实战三部曲》的终篇,次要内容是在 CoProcessFunction 中应用定时器和侧输入,对上一篇的性能进行加强;
  • 回顾上一篇的性能:一号流收到 <font color=”blue”>aaa</font> 后保留在状态中,直到二号流收到 aaa,把两个 aaa 的值相加后输入到上游;
  • 上述性能有个问题:二号流如果始终收不到 <font color=”blue”>aaa</font>,上游就始终没有 aaa 的输入,相当于进入一号流的 aaa 曾经杳无音信了;
  • 明天的实战就是修复上述问题:aaa 在一个流中呈现后,10 秒之内如果呈现在另一个流中,就像以前那样值相加,输入到上游,如果 10 秒内没有呈现在另一个流,就流向侧输入,再将所有状态清理洁净;

    参考文章

  • 了解状态:《深刻理解 ProcessFunction 的状态操作(Flink-1.10)》
  • 了解定时器:《了解 ProcessFunction 的 Timer 逻辑》

    梳理流程

  • 为了编码的逻辑正确,咱们把失常和异样的流程先梳理分明;
  • 下图是失常流程:aaa 在一号流呈现后,10 秒内又在二号流呈现了,于是相加并流向上游:

  • 再来看异样的流程,如下图,一号流在 <font color=”blue”>16:14:01</font> 收到 aaa,但二号流始终没有收到 aaa,等到 10 秒后,也就是 <font color=”blue”>16:14:11</font>,定时器被触发,从状态 1 得悉 10 秒前一号流收到过 aaa,于是将数据流向一号侧输入:

  • 接下来编码实现下面的性能;

    源码下载

    如果您不想写代码,整个系列的源码可在 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> 文件夹下,如下图红框所示:

CoProcessFunction 的子类

  1. 后面的两篇实战中,<font color=”blue”>CoProcessFunction</font> 的子类都写成了匿名类,如下图红框:

  1. 本文中,CoProcessFunction 子类会用到外部类的成员变量,因而不能再用匿名类了,新增 CoProcessFunction 的子类 <font color=”blue”>ExecuteWithTimeoutCoProcessFunction.java</font>,稍后会阐明几个关键点:
package com.bolingcavalry.coprocessfunction;

import com.bolingcavalry.Utils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 实现双流业务逻辑的性能类
 */
public class ExecuteWithTimeoutCoProcessFunction extends CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> {private static final Logger logger = LoggerFactory.getLogger(ExecuteWithTimeoutCoProcessFunction.class);

    /**
     * 等待时间
     */
    private static final long WAIT_TIME = 10000L;

    public ExecuteWithTimeoutCoProcessFunction(OutputTag<String> source1SideOutput, OutputTag<String> source2SideOutput) {super();
        this.source1SideOutput = source1SideOutput;
        this.source2SideOutput = source2SideOutput;
    }

    private OutputTag<String> source1SideOutput;

    private OutputTag<String> source2SideOutput;

    // 某个 key 在 processElement1 中存入的状态
    private ValueState<Integer> state1;

    // 某个 key 在 processElement2 中存入的状态
    private ValueState<Integer> state2;

    // 如果创立了定时器,就在状态中保留定时器的 key
    private ValueState<Long> timerState;

    // onTimer 中拿不到以后 key,只能提前保留在状态中(KeyedProcessFunction 的 OnTimerContext 有 API 能够取到,然而 CoProcessFunction 的 OnTimerContext 却没有)private ValueState<String> currentKeyState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化状态
        state1 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState1", Integer.class));
        state2 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState2", Integer.class));
        timerState = getRuntimeContext().getState(new ValueStateDescriptor<>("timerState", Long.class));
        currentKeyState = getRuntimeContext().getState(new ValueStateDescriptor<>("currentKeyState", String.class));
    }

    /**
     * 所有状态都清理掉
     */
    private void clearAllState() {state1.clear();
        state2.clear();
        currentKeyState.clear();
        timerState.clear();}

    @Override
    public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {logger.info("processElement1:解决元素 1:{}", value);

        String key = value.f0;

        Integer value2 = state2.value();

        // value2 为空,就示意 processElement2 还没有解决或这个 key,// 这时候就把 value1 保存起来
        if(null==value2) {logger.info("processElement1:2 号流还未收到过 [{}],把 1 号流收到的值[{}] 保存起来", key, value.f1);
            state1.update(value.f1);

            currentKeyState.update(key);

            // 开始 10 秒的定时器,10 秒后会进入
            long timerKey = ctx.timestamp() + WAIT_TIME;
            ctx.timerService().registerProcessingTimeTimer(timerKey);
            // 保留定时器的 key
            timerState.update(timerKey);
            logger.info("processElement1:创立定时器[{}],期待 2 号流接收数据", Utils.time(timerKey));
        } else {logger.info("processElement1:2 号流收到过[{}],值是[{}],当初把两个值相加后输入", key, value2);

            // 输入一个新的元素到上游节点
            out.collect(new Tuple2<>(key, value.f1 + value2));

            // 删除定时器(这个定时器应该是 processElement2 创立的)long timerKey = timerState.value();
            logger.info("processElement1:[{}]的新元素已输入到上游,删除定时器[{}]", key, Utils.time(timerKey));
            ctx.timerService().deleteProcessingTimeTimer(timerKey);

            clearAllState();}
    }

    @Override
    public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {logger.info("processElement2:解决元素 2:{}", value);

        String key = value.f0;

        Integer value1 = state1.value();

        // value1 为空,就示意 processElement1 还没有解决或这个 key,// 这时候就把 value2 保存起来
        if(null==value1) {logger.info("processElement2:1 号流还未收到过 [{}],把 2 号流收到的值[{}] 保存起来", key, value.f1);
            state2.update(value.f1);

            currentKeyState.update(key);

            // 开始 10 秒的定时器,10 秒后会进入
            long timerKey = ctx.timestamp() + WAIT_TIME;
            ctx.timerService().registerProcessingTimeTimer(timerKey);
            // 保留定时器的 key
            timerState.update(timerKey);
            logger.info("processElement2:创立定时器[{}],期待 1 号流接收数据", Utils.time(timerKey));
        } else {logger.info("processElement2:1 号流收到过[{}],值是[{}],当初把两个值相加后输入", key, value1);

            // 输入一个新的元素到上游节点
            out.collect(new Tuple2<>(key, value.f1 + value1));

            // 删除定时器(这个定时器应该是 processElement1 创立的)long timerKey = timerState.value();
            logger.info("processElement2:[{}]的新元素已输入到上游,删除定时器[{}]", key, Utils.time(timerKey));
            ctx.timerService().deleteProcessingTimeTimer(timerKey);

            clearAllState();}
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {super.onTimer(timestamp, ctx, out);

        String key = currentKeyState.value();

        // 定时器被触发,意味着此 key 只在一个中呈现过
        logger.info("[{}]的定时器 [{}] 被触发了", key, Utils.time(timestamp));

        Integer value1 = state1.value();
        Integer value2 = state2.value();

        if(null!=value1) {logger.info("只有 1 号流收到过[{}],值为[{}]", key, value1);
            // 侧输入
            ctx.output(source1SideOutput, "source1 side, key [" + key+ "], value [" + value1 + "]");
        }

        if(null!=value2) {logger.info("只有 2 号流收到过[{}],值为[{}]", key, value2);
            // 侧输入
            ctx.output(source2SideOutput, "source2 side, key [" + key+ "], value [" + value2 + "]");
        }

        clearAllState();}
}
  1. 关键点之一:新增状态 <font color=”blue”>timerState</font>,用于保留定时器的 key;
  2. 关键点之二:CoProcessFunction 的 <font color=”blue”>onTimer</font> 中拿不到以后 key(KeyedProcessFunction 能够,其 OnTimerContext 类提供了 API),因而新增状态 <font color=”blue”>currentKeyState</font>,这样在 onTimer 中就晓得以后 key 了;
  3. 关键点之三:processElement1 中,解决 aaa 时,如果 2 号流还没收到过 aaa,就存入状态,并启动 10 秒定时器;
  4. 关键点之四:processElement2 解决 aaa 时,发现 1 号流收到过 aaa,就相加再输入到上游,并且删除 processElement1 中创立的定时器,aaa 相干的所有状态也全副清理掉;
  5. 关键点之五:如果 10 秒内 aaa 在两个流中都呈现过,那么肯定会流入上游并且定时器会被删除,因而,一旦 <font color=”blue”>onTimer</font> 被执行,意味着 aaa 只在一个流中呈现过,而且曾经过来 10 秒了,此时在 <font color=”blue”>onTimer</font> 中能够执行流向侧输入的操作;
  6. 以上就是双流解决的逻辑和代码,接下来编写 <font color=”blue”>AbstractCoProcessFunctionExecutor</font> 的子类;

业务执行类 AddTwoSourceValueWithTimeout

  1. 负责执行整个性能的,是抽象类 <font color=”blue”>AbstractCoProcessFunctionExecutor</font> 的子类,如下,稍后会阐明几个关键点:
package com.bolingcavalry.coprocessfunction;

import com.bolingcavalry.Utils;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
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.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author will
 * @email zq2599@gmail.com
 * @date 2020-11-11 09:48
 * @description 将两个流中相通 key 的 value 相加,当 key 在一个流中呈现后,*              会在无限工夫内期待它在另一个流中呈现,如果超过等待时间任未呈现就在旁路输入
 */
public class AddTwoSourceValueWithTimeout extends AbstractCoProcessFunctionExecutor {private static final Logger logger = LoggerFactory.getLogger(AddTwoSourceValueWithTimeout.class);

    // 假如 aaa 流入 1 号源后,在 2 号源超过 10 秒没有收到 aaa,那么 1 号源的 aaa 就会流入 source1SideOutput
    final OutputTag<String> source1SideOutput = new OutputTag<String>("source1-sideoutput"){};

    // 假如 aaa 流入 2 号源后,如果 1 号源超过 10 秒没有收到 aaa,那么 2 号源的 aaa 就会流入 source2SideOutput
    final OutputTag<String> source2SideOutput = new OutputTag<String>("source2-sideoutput"){};

    /**
     * 重写父类的办法,放弃父类逻辑不变,仅减少了工夫戳分配器,向元素中退出工夫戳
     * @param port
     * @return
     */
    @Override
    protected KeyedStream<Tuple2<String, Integer>, Tuple> buildStreamFromSocket(StreamExecutionEnvironment env, int port) {
        return env
                // 监听端口
                .socketTextStream("localhost", port)
                // 失去的字符串 "aaa,3" 转成 Tuple2 实例,f0="aaa",f1=3
                .map(new WordCountMap())
                // 设置工夫戳分配器,用以后工夫作为工夫戳
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {

                    @Override
                    public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {long timestamp = System.currentTimeMillis();
                        logger.info("增加工夫戳,值:{},工夫戳:{}", element, Utils.time(timestamp));
                        // 应用以后零碎工夫作为工夫戳
                        return timestamp;
                    }

                    @Override
                    public Watermark getCurrentWatermark() {
                        // 本例不须要 watermark,返回 null
                        return null;
                    }
                })
                // 将单词作为 key 分区
                .keyBy(0);
    }

    @Override
    protected CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> getCoProcessFunctionInstance() {return new ExecuteWithTimeoutCoProcessFunction(source1SideOutput, source2SideOutput);
    }

    @Override
    protected void doSideOutput(SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream) {
        // 两个侧输入都间接打印
        mainDataStream.getSideOutput(source1SideOutput).print();
        mainDataStream.getSideOutput(source2SideOutput).print();}

    public static void main(String[] args) throws Exception {new AddTwoSourceValueWithTimeout().execute();}
}
  1. 关键点之一:增减成员变量 <font color=”blue”>source1SideOutput</font> 和 <font color=”blue”>source2SideOutput</font>,用于侧输入;
  2. 关键点之二:重写父类的 buildStreamFromSocket 办法,加了个工夫戳分配器,这样每个元素都带有工夫戳;
  3. 关键点之三:重写父类的 doSideOutput 办法,这外面会把侧输入的数据打印进去;
  4. 以上就是所有代码了,接下来开始验证;

验证(不超时的操作)

  1. 别离开启本机的 <font color=”blue”>9998</font> 和 <font color=”blue”>9999</font> 端口,我这里是 MacBook,执行 <font color=”blue”>nc -l 9998</font> 和 <font color=”blue”>nc -l 9999</font>
  2. 启动 Flink 利用,如果您和我一样是 Mac 电脑,间接运行 <font color=”blue”>AddTwoSourceValueWithTimeout.main</font> 办法即可(如果是 windows 电脑,我这没试过,不过做成 jar 在线部署也是能够的);
  3. 在监听 9998 端口的控制台输出 <font color=”blue”>aaa,1</font>,此时 flink 控制台输入如下,可见 processElement1 办法中,读取 state2 为空,示意 aaa 在 2 号流还未呈现过,此时的 aaa 是首次呈现,应该放入 state 中保留,并且创立了定时器:
18:18:10,472 INFO  AddTwoSourceValueWithTimeout  - 增加工夫戳,值:(aaa,1),工夫戳:2020-11-12 06:18:10
18:18:10,550 INFO  ExecuteWithTimeoutCoProcessFunction  - processElement1:解决元素 1:(aaa,1)
18:18:10,550 INFO  ExecuteWithTimeoutCoProcessFunction  - processElement1:2 号流还未收到过 [aaa],把 1 号流收到的值[1] 保存起来
18:18:10,553 INFO  ExecuteWithTimeoutCoProcessFunction  - processElement1:创立定时器[2020-11-12 06:18:20],期待 2 号流接收数据
  1. 尽快在监听 9999 端口的控制台输出 <font color=”blue”>aaa,2</font>,flink 日志如下所示,可见相加后输入到上游,并且定时器也删除了:
18:18:15,813 INFO  AddTwoSourceValueWithTimeout  - 增加工夫戳,值:(aaa,2),工夫戳:2020-11-12 06:18:15
18:18:15,887 INFO  ExecuteWithTimeoutCoProcessFunction  - processElement2:解决元素 2:(aaa,2)
18:18:15,887 INFO  ExecuteWithTimeoutCoProcessFunction  - processElement2:1 号流收到过[aaa],值是[1],当初把两个值相加后输入
(aaa,3)
18:18:15,888 INFO  ExecuteWithTimeoutCoProcessFunction  - processElement2:[aaa]的新元素已输入到上游,删除定时器[2020-11-12 06:18:20]

验证(超时的操作)

  1. 前面试过了失常流程,再来试试超时流程是否合乎预期;
  2. 在监听 9998 端口的控制台输出 <font color=”blue”>aaa,1</font>,而后期待十秒,flink 控制台输入如下,可见定时器被触发,并且 aaa 流向了 1 号流的侧输入:
18:23:37,393 INFO  AddTwoSourceValueWithTimeout - 增加工夫戳,值:(aaa,1),工夫戳:2020-11-12 06:23:37
18:23:37,417 INFO  ExecuteWithTimeoutCoProcessFunction - processElement1:解决元素 1:(aaa,1)
18:23:37,417 INFO  ExecuteWithTimeoutCoProcessFunction - processElement1:2 号流还未收到过 [aaa],把 1 号流收到的值[1] 保存起来
18:23:37,417 INFO  ExecuteWithTimeoutCoProcessFunction - processElement1:创立定时器[2020-11-12 06:23:47],期待 2 号流接收数据
18:23:47,398 INFO  ExecuteWithTimeoutCoProcessFunction - [aaa]的定时器 [2020-11-12 06:23:47] 被触发了
18:23:47,399 INFO  ExecuteWithTimeoutCoProcessFunction - 只有 1 号流收到过[aaa],值为[1]
source1 side, key [aaa], value [1]
  • 至此,CoProcessFunction 实战三部曲曾经全副实现了,心愿这三次实战可能给您一些参考,帮您更快把握和了解 CoProcessFunction;

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

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

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

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

退出移动版