CoProcessFunction实战三部曲之三：定时器和侧输出，linux容器技术

2022-04-07 09:50:35

系列文章链接

  1. 基本功能

  2. 状态处理

  3. 定时器和侧输出

本篇概览

  • 本文是《CoProcessFunction实战三部曲》的终篇，主要内容是在CoProcessFunction中使用定时器和侧输出，对上一篇的功能进行增强；

  • 回顾上一篇的功能：一号流收到aaa后保存在状态中，直到二号流收到aaa，把两个aaa的值相加后输出到下游；

  • 上述功能有个问题：二号流如果一直收不到aaa，下游就一直没有aaa的输出，相当于进入一号流的aaa已经石沉大海了；

  • 今天的实战就是修复上述问题：aaa在一个流中出现后，10秒之内如果出现在另一个流中，就像以前那样值相加，输出到下游，如果10秒内没有出现在另一个流，就流向侧输出，再将所有状态清理干净；

参考文章

  1. 理解状态：《深入了解ProcessFunction的状态操作(Flink-1.10)》

  2. 理解定时器：《理解ProcessFunction的Timer逻辑》

梳理流程

  • 为了编码的逻辑正确，咱们把正常和异常的流程先梳理清楚；

  • 下图是正常流程：aaa在一号流出现后，10秒内又在二号流出现了，于是相加并流向下游：

CoProcessFunction实战三部曲之三：定时器和侧输出，linux容器技术

  • 再来看异常的流程，如下图，一号流在16:14:01收到aaa，但二号流一直没有收到aaa，等到10秒后，也就是16:14:11，定时器被触发，从状态1得知10秒前一号流收到过aaa，于是将数据流向一号侧输出：

CoProcessFunction实战三部曲之三：定时器和侧输出，linux容器技术

  • 接下来编码实现上面的功能；

源码下载

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

| 名称 | 链接 | 备注 |

| :-- | :-- | :-- |

| 项目主页 | https://github.com/zq2599/blog_demos | 该项目在GitHub上的主页 |

| git仓库地址(https) | https://github.com/zq2599/blog_demos.git | 该项目源码的仓库地址，https协议 |

| git仓库地址(ssh) | git@github.com:zq2599/blog_demos.git | 该项目源码的仓库地址，ssh协议 |

这个git项目中有多个文件夹，本章的应用在flinkstudy文件夹下，如下图红框所示：

CoProcessFunction实战三部曲之三：定时器和侧输出，linux容器技术

CoProcessFunction的子类

  1. 前面的两篇实战中，CoProcessFunction的子类都写成了匿名类，如下图红框：

CoProcessFunction实战三部曲之三：定时器和侧输出，linux容器技术

  1. 本文中，CoProcessFunction子类会用到外部类的成员变量，因此不能再用匿名类了，新增CoProcessFunction的子类ExecuteWithTimeoutCoProcessFunction.java，稍后会说明几个关键点：

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 source1SideOutput, OutputTag source2SideOutput) {

super();

this.source1SideOutput = source1SideOutput;

this.source2SideOutput = source2SideOutput;

}

private OutputTag source1SideOutput;

private OutputTag source2SideOutput;

// 某个key在processElement1中存入的状态

private ValueState state1;

// 某个key在processElement2中存入的状态

private ValueState state2;

// 如果创建了定时器，就在状态中保存定时器的key

private ValueState timerState;

// onTimer中拿不到当前key，只能提前保存在状态中（KeyedProcessFunction的OnTimerContext有API可以取到，但是CoProcessFunction的OnTimerContext却没有）

private ValueState 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. 关键点之一：新增状态timerState，用于保存定时器的key；

  2. 关键点之二：CoProcessFunction的onTimer中拿不到当前key(KeyedProcessFunction可以，其OnTimerContext类提供了API)，因此新增状态currentKeyState，这样在onTimer中就知道当前key了；

  3. 关键点之三：processElement1中，处理aaa时， 如果2号流还没收到过aaa，就存入状态，并启动10秒定时器；

  4. 关键点之四：processElement2处理aaa时，发现1号流收到过aaa，就相加再输出到下游，并且删除processElement1中创建的定时器，aaa相关的所有状态也全部清理掉；

  5. 关键点之五：如果10秒内aaa在两个流中都出现过，那么一定会流入下游并且定时器会被删除，因此，一旦onTimer被执行，意味着aaa只在一个流中出现过，而且已经过去10秒了，此时在onTimer中可以执行流向侧输出的操作；

  6. 以上就是双流处理的逻辑和代码，接下来编写AbstractCoProcessFunctionExecutor的子类；

业务执行类AddTwoSourceValueWithTimeout

  1. 负责执行整个功能的，是抽象类AbstractCoProcessFunctionExecutor的子类，如下，稍后会说明几个关键点：

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);

上一篇：大家信夫诚信认证条件，简约而不简单


下一篇：JS的基础以及对与HTML和CSS这三者的理解