CoProcessFunction实战三部曲之二：状态处理

àì夳堔傛蜴生んèń 2022-12-20 09:05 97阅读 0赞

### 欢迎访问我的GitHub ###

> 这里分类和汇总了欣宸的全部原创(含配套源码)：[https://github.com/zq2599/blog\_demos][https_github.com_zq2599_blog_demos]

### 系列文章链接 ###

1.  [基本功能][Link 1]
2.  [状态处理][Link 2]
3.  [定时器和侧输出][Link 3]

### 本篇概览 ###

*  本文是《CoProcessFunction实战三部曲》的第二篇，咱们要实战的是双流连接场景下，处理一号流中的数据时，还要结合该key在二号流中的情况；
 *  最简单的例子：aaa在一号流中的value和二号流的value相加，再输出到下游，如下图所示，一号流中的value存入state，在二号流中取出并相加，将结果输出给下游：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center]
 *  本篇的内容就是编码实现上图的功能；

### 参考文章 ###

理解状态：[《深入了解ProcessFunction的状态操作(Flink-1.10)》][ProcessFunction_Flink-1.10]

### 源码下载 ###

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

<table> 
 <thead> 
  <tr> 
   <th align="left">名称</th> 
   <th align="left">链接</th> 
   <th align="left">备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">项目主页</td> 
   <td align="left">https://github.com/zq2599/blog_demos</td> 
   <td align="left">该项目在GitHub上的主页</td> 
  </tr> 
  <tr> 
   <td align="left">git仓库地址(https)</td> 
   <td align="left">https://github.com/zq2599/blog_demos.git</td> 
   <td align="left">该项目源码的仓库地址，https协议</td> 
  </tr> 
  <tr> 
   <td align="left">git仓库地址(ssh)</td> 
   <td align="left">git@github.com:zq2599/blog_demos.git</td> 
   <td align="left">该项目源码的仓库地址，ssh协议</td> 
  </tr> 
 </tbody> 
</table>

这个git项目中有多个文件夹，本章的应用在flinkstudy文件夹下，如下图红框所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70]

### 编码 ###

1.  字符串转Tuple2的Map函数，以及抽象类AbstractCoProcessFunctionExecutor都和上一篇[《CoProcessFunction实战三部曲之一：基本功能》][Link 1]一模一样；
2.  新增AbstractCoProcessFunctionExecutor的子类AddTwoSourceValue.java，源码如下，稍后会说明几个关键点：

package com.bolingcavalry.coprocessfunction;
    
    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.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /** * @author will * @email zq2599@gmail.com * @date 2020-11-11 09:48 * @description 功能介绍 */
    public class AddTwoSourceValue extends AbstractCoProcessFunctionExecutor { 
    
        private static final Logger logger = LoggerFactory.getLogger(AddTwoSourceValue.class);
    
        @Override
        protected CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> getCoProcessFunctionInstance() { 
            return new CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>() { 
    
                // 某个key在processElement1中存入的状态
                private ValueState<Integer> state1;
    
                // 某个key在processElement2中存入的状态
                private ValueState<Integer> state2;
    
                @Override
                public void open(Configuration parameters) throws Exception { 
                    // 初始化状态
                    state1 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState1", Integer.class));
                    state2 = getRuntimeContext().getState(new ValueStateDescriptor<>("myState2", Integer.class));
                }
    
                @Override
                public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { 
                    logger.info("处理元素1：{}", value);
    
                    String key = value.f0;
    
                    Integer value2 = state2.value();
    
                    // value2为空，就表示processElement2还没有处理或这个key，
                    // 这时候就把value1保存起来
                    if(null==value2) { 
                        logger.info("2号流还未收到过[{}]，把1号流收到的值[{}]保存起来", key, value.f1);
                        state1.update(value.f1);
                    } else { 
                        logger.info("2号流收到过[{}]，值是[{}]，现在把两个值相加后输出", key, value2);
    
                        // 输出一个新的元素到下游节点
                        out.collect(new Tuple2<>(key, value.f1 + value2));
    
                        // 把2号流的状态清理掉
                        state2.clear();
                    }
                }
    
                @Override
                public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception { 
                    logger.info("处理元素2：{}", value);
    
                    String key = value.f0;
    
                    Integer value1 = state1.value();
    
                    // value1为空，就表示processElement1还没有处理或这个key，
                    // 这时候就把value2保存起来
                    if(null==value1) { 
                        logger.info("1号流还未收到过[{}]，把2号流收到的值[{}]保存起来", key, value.f1);
                        state2.update(value.f1);
                    } else { 
                        logger.info("1号流收到过[{}]，值是[{}]，现在把两个值相加后输出", key, value1);
    
                        // 输出一个新的元素到下游节点
                        out.collect(new Tuple2<>(key, value.f1 + value1));
    
                        // 把1号流的状态清理掉
                        state1.clear();
                    }
                }
            };
        }
    
        public static void main(String[] args) throws Exception { 
            new AddTwoSourceValue().execute();
        }
    }

1.  关键点之一：对于aaa这个key，无法确定会先出现在一号源还是二号源，如果先出现在一号源，就应该在processElement1中将value保存在state1中，这样等到aaa再次出现在二号源时，processElement2就可以从state1中取出一号源的value，相加后输出到下游；
2.  关键点之二：如果输出到下游，就表示数据已经处理完毕，此时要把保存的状态清理掉；
3.  如果您想了解低阶函数中的状态存取的更多细节，请参考[《深入了解ProcessFunction的状态操作(Flink-1.10)》][ProcessFunction_Flink-1.10]

### 验证 ###

1.  分别开启本机的9998和9999端口，我这里是MacBook，执行nc -l 9998和nc -l 9999
2.  启动Flink应用，如果您和我一样是Mac电脑，直接运行AddTwoSourceValue.main方法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是可以的）；
3.  在监听9998端口的控制台输入aaa,111，此时flink控制台输出如下，可见processElement1方法中，读取state2为空，表示aaa在二号流还未出现过，此时的aaa是首次出现，应该放入state中保存：

22:35:12,135 INFO  AddTwoSourceValue - 处理元素1：(aaa,111)
    22:35:12,136 INFO  AddTwoSourceValue - 2号流还未收到过[aaa]，把1号流收到的值[111]保存起来

1.  在监听9999端口的控制台输入bbb,123，flink日志如下所示，表示bbb也是首次出现，把值保存在state中：

22:35:34,473 INFO  AddTwoSourceValue - 处理元素2：(bbb,123)
    22:35:34,473 INFO  AddTwoSourceValue - 1号流还未收到过[bbb]，把2号流收到的值[123]保存起来

1.  在监听9999端口的控制台输入aaa,222，flink日志如下，很明显，之前保存在state中的值被取出来了，因此processElement2方法中，aaa在两个数据源的值111和222会被相加后输出到下游，下游是print，直接打印出来了：

22:35:38,072 INFO  AddTwoSourceValue - 处理元素2：(aaa,222)
    22:35:38,072 INFO  AddTwoSourceValue - 1号流收到过[aaa]，值是[111]，现在把两个值相加后输出
    (aaa,333)

*  至此，双流场景下的状态互通实践咱们已经完成了，接下来的文章，会加上定时器和旁路输出，将双流场景的数据处理考虑得更加全面；

### 欢迎关注公众号：程序员欣宸 ###

> 微信搜索「程序员欣宸」，我是欣宸，期待与您一同畅游Java世界…

[https_github.com_zq2599_blog_demos]: https://github.com/zq2599/blog_demos
[Link 1]: https://xinchen.blog.csdn.net/article/details/109624375
[Link 2]: https://xinchen.blog.csdn.net/article/details/109629119
[Link 3]: https://xinchen.blog.csdn.net/article/details/109645214
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/e64237e08ae94aaf9b308b3cd338edc9.png
[ProcessFunction_Flink-1.10]: https://xinchen.blog.csdn.net/article/details/106040312
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JvbGluZ19jYXZhbHJ5_size_16_color_FFFFFF_t_70]: /images/20221120/20210b694f5c49bfba6511f505437dd8.png