CoProcessFunction实战三部曲之三：定时器和侧输出

Posted 2022-04-12 程序员欣宸

欢迎访问我的GitHub

本篇概览

• 本文是《CoProcessFunction实战三部曲》的终篇，主要内容是在CoProcessFunction中使用定时器和侧输出，对上一篇的功能进行增强；
• 回顾上一篇的功能：一号流收到aaa后保存在状态中，直到二号流收到aaa，把两个aaa的值相加后输出到下游；
• 上述功能有个问题：二号流如果一直收不到aaa，下游就一直没有aaa的输出，相当于进入一号流的aaa已经石沉大海了；
• 今天的实战就是修复上述问题：aaa在一个流中出现后，10秒之内如果出现在另一个流中，就像以前那样值相加，输出到下游，如果10秒内没有出现在另一个流，就流向侧输出，再将所有状态清理干净；

  参考文章

  1. 理解状态：《深入了解ProcessFunction的状态操作(Flink-1.10)》
  2. 理解定时器：《理解ProcessFunction的Timer逻辑》

     梳理流程

• 为了编码的逻辑正确，咱们把正常和异常的流程先梳理清楚；
• 下图是正常流程：aaa在一号流出现后，10秒内又在二号流出现了，于是相加并流向下游：
  
• 再来看异常的流程，如下图，一号流在16:14:01收到aaa，但二号流一直没有收到aaa，等到10秒后，也就是16:14:11，定时器被触发，从状态1得知10秒前一号流收到过aaa，于是将数据流向一号侧输出：
  
• 接下来编码实现上面的功能；

  源码下载

  如果您不想写代码，整个系列的源码可在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)	[e-mail]:zq2599/blog_demos.git	该项目源码的仓库地址，ssh协议

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

CoProcessFunction的子类

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

2. 本文中，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<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();

  
  

  
  3. 关键点之一：新增状态**timerState**，用于保存定时器的key；
  4. 关键点之二：CoProcessFunction的**onTimer**中拿不到当前key(KeyedProcessFunction可以，其OnTimerContext类提供了API)，因此新增状态**currentKeyState**，这样在onTimer中就知道当前key了；
  5. 关键点之三：processElement1中，处理aaa时， 如果2号流还没收到过aaa，就存入状态，并启动10秒定时器；
  6. 关键点之四：processElement2处理aaa时，发现1号流收到过aaa，就相加再输出到下游，并且删除processElement1中创建的定时器，aaa相关的所有状态也全部清理掉；
  7. 关键点之五：如果10秒内aaa在两个流中都出现过，那么一定会流入下游并且定时器会被删除，因此，一旦**onTimer**被执行，意味着aaa只在一个流中出现过，而且已经过去10秒了，此时在**onTimer**中可以执行流向侧输出的操作；
  8. 以上就是双流处理的逻辑和代码，接下来编写**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 [e-mail]
• @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

  • @returnbr/>*/
    @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();
  
  

  
  2. 关键点之一：增减成员变量**source1SideOutput**和**source2SideOutput**，用于侧输出；
  3. 关键点之二：重写父类的buildStreamFromSocket方法，加了个时间戳分配器，这样每个元素都带有时间戳；
  4. 关键点之三：重写父类的doSideOutput方法，这里面会把侧输出的数据打印出来；
  5. 以上就是所有代码了，接下来开始验证；

验证（不超时的操作）

1. 分别开启本机的9998和9999端口，我这里是MacBook，执行nc -l 9998和nc -l 9999
2. 启动Flink应用，如果您和我一样是Mac电脑，直接运行AddTwoSourceValueWithTimeout.main方法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是可以的）；
3. 在监听9998端口的控制台输入aaa,1，此时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号流接收数据

4. 尽快在监听9999端口的控制台输入aaa,2，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]

   验证（超时的操作）

5. 前面试过了正常流程，再来试试超时流程是否符合预期；
6. 在监听9998端口的控制台输入aaa,1，然后等待十秒，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；

欢迎关注51CTO博客：程序员欣宸