Flink处理函数实战之五：CoProcessFunction(双流处理)-低调大师

Flink处理函数实战之五：CoProcessFunction(双流处理)

2020-11-23 537

欢迎访问我的GitHub

https://github.com/zq2599/blog_demos

内容：所有原创文章分类汇总及配套源码，涉及Java、Docker、Kubernetes、DevOPS等；

欢迎访问我的GitHub

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

Flink处理函数实战系列链接

本篇概览

本文是《Flink处理函数实战》系列的第五篇，学习内容是如何同时处理两个数据源的数据；
试想在面对两个输入流时，如果这两个流的数据之间有业务关系，该如何编码实现呢，例如下图中的操作，同时监听9998和9999端口，将收到的输出分别处理后，再由同一个sink处理(打印)：
Flink支持的方式是扩展CoProcessFunction来处理，为了更清楚认识，我们把KeyedProcessFunction和CoProcessFunction的类图摆在一起看，如下所示：
从上图可见，CoProcessFunction和KeyedProcessFunction的继承关系一样，另外CoProcessFunction自身也很简单，在processElement1和processElement2中分别处理两个上游流入的数据即可，并且也支持定时器设置；

编码实战

接下来咱们开发一个应用来体验CoProcessFunction，功能非常简单，描述如下：

建两个数据源，数据分别来自本地9998和9999端口；
每个端口收到类似aaa,123这样的数据，转成Tuple2实例，f0是aaa，f1是123；
在CoProcessFunction的实现类中，对每个数据源的数据都打日志，然后全部传到下游算子；
下游操作是打印，因此9998和9999端口收到的所有数据都会在控制台打印出来；
整个demo的功能如下图所示：

接下来编码实现上述功能；

源码下载

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

Map算子

做一个map算子，用来将字符串aaa,123转成Tuple2实例，f0是aaa，f1是123；
算子名为WordCountMap.java：

package com.bolingcavalry.coprocessfunction;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.StringUtils;

public class WordCountMap implements MapFunction<String, Tuple2<String, Integer>> {
    @Override
    public Tuple2<String, Integer> map(String s) throws Exception {

        if(StringUtils.isNullOrWhitespaceOnly(s)) {
            System.out.println("invalid line");
            return null;
        }

        String[] array = s.split(",");

        if(null==array || array.length<2) {
            System.out.println("invalid line for array");
            return null;
        }

        return new Tuple2<>(array[0], Integer.valueOf(array[1]));
    }
}

便于扩展的抽象类

开发一个抽象类，将前面图中提到的监听端口、map处理、keyby处理、打印都做到这个抽象类中，但是CoProcessFunction的逻辑却不放在这里，而是交给子类来实现，这样如果我们想进一步实践和扩展CoProcessFunction的能力，只要在子类中专注做好CoProcessFunction相关开发即可，如下图，红色部分交给子类实现，其余的都是抽象类完成的：
抽象类AbstractCoProcessFunctionExecutor.java，源码如下，稍后会说明几个关键点：

package com.bolingcavalry.coprocessfunction;

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.co.CoProcessFunction;

/**
 * @author will
 * @email zq2599@gmail.com
 * @date 2020-11-09 17:33
 * @description 串起整个逻辑的执行类，用于体验CoProcessFunction
 */
public abstract class AbstractCoProcessFunctionExecutor {

    /**
     * 返回CoProcessFunction的实例，这个方法留给子类实现
     * @return
     */
    protected abstract CoProcessFunction<
            Tuple2<String, Integer>,
            Tuple2<String, Integer>,
            Tuple2<String, Integer>> getCoProcessFunctionInstance();

    /**
     * 监听根据指定的端口，
     * 得到的数据先通过map转为Tuple2实例，
     * 给元素加入时间戳，
     * 再按f0字段分区，
     * 将分区后的KeyedStream返回
     * @param port
     * @return
     */
    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())
                // 将单词作为key分区
                .keyBy(0);
    }

    /**
     * 如果子类有侧输出需要处理，请重写此方法，会在主流程执行完毕后被调用
     */
    protected void doSideOutput(SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream) {
    }

    /**
     * 执行业务的方法
     * @throws Exception
     */
    public void execute() throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 并行度1
        env.setParallelism(1);

        // 监听9998端口的输入
        KeyedStream<Tuple2<String, Integer>, Tuple> stream1 = buildStreamFromSocket(env, 9998);

        // 监听9999端口的输入
        KeyedStream<Tuple2<String, Integer>, Tuple> stream2 = buildStreamFromSocket(env, 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream = stream1
                // 两个流连接
                .connect(stream2)
                // 执行低阶处理函数，具体处理逻辑在子类中实现
                .process(getCoProcessFunctionInstance());

        // 将低阶处理函数输出的元素全部打印出来
        mainDataStream.print();

        // 侧输出相关逻辑，子类有侧输出需求时重写此方法
        doSideOutput(mainDataStream);

        // 执行
        env.execute("ProcessFunction demo : CoProcessFunction");
    }
}

关键点之一：一共有两个数据源，每个源的处理逻辑都封装到buildStreamFromSocket方法中；
关键点之二：stream1.connect(stream2)将两个流连接起来；
关键点之三：process接收CoProcessFunction实例，合并后的流的处理逻辑就在这里面；
关键点之四：getCoProcessFunctionInstance是抽象方法，返回CoProcessFunction实例，交给子类实现，所以CoProcessFunction中做什么事情完全由子类决定；
关键点之五：doSideOutput方法中啥也没做，但是在主流程代码的末尾会被调用，如果子类有侧输出(SideOutput)的需求，重写此方法即可，此方法的入参是处理过的数据集，可以从这里取得侧输出；

子类决定CoProcessFunction的功能

子类CollectEveryOne.java如下所示，逻辑很简单，将每个源的上游数据直接输出到下游算子：

package com.bolingcavalry.coprocessfunction;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CollectEveryOne extends AbstractCoProcessFunctionExecutor {

    private static final Logger logger = LoggerFactory.getLogger(CollectEveryOne.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>>() {

            @Override
            public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) {
                logger.info("处理1号流的元素：{},", value);
                out.collect(value);
            }

            @Override
            public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) {
                logger.info("处理2号流的元素：{}", value);
                out.collect(value);
            }
        };
    }

    public static void main(String[] args) throws Exception {
        new CollectEveryOne().execute();
    }
}

上述代码中，CoProcessFunction后面的泛型定义很长：<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> ，一共三个Tuple2，分别代表一号数据源输入、二号数据源输入、下游输出的类型；

验证

分别开启本机的9998和9999端口，我这里是MacBook，执行nc -l 9998和nc -l 9999
启动Flink应用，如果您和我一样是Mac电脑，直接运行CollectEveryOne.main方法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是可以的）；
在监听9998和9999端口的控制台分别输入aaa,111和bbb,222
以下是flink控制台输出的内容，可见processElement1和processElement1方法的日志代码已经执行，并且print方法作为最下游，将两个数据源的数据都打印出来了，符合预期：

12:45:38,774 INFO CollectEveryOne - 处理1号流的元素：(aaa,111),
(aaa,111)
12:45:43,816 INFO CollectEveryOne - 处理2号流的元素：(bbb,222)
(bbb,222)

你不孤单，欣宸原创一路相伴

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

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

微信关注我们

原文链接：https://my.oschina.net/zq2599/blog/4740268

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

我的亲历：一行代码，百万人民币打水漂

导读：几年前，我刚进入职场，作为程序员走上了技术这条路，不久便亲身经历了一件特别震撼的事情。那是一行代码引发的线上故障，故障造成了百万级的资金损失。时至今日，我依然印象深刻，也正是这件事，让我在职业生涯初期就形成了敬畏代码，严谨做事的态度。时间过去的比较久，我脑海里的细节也存在失真。我将尽量还原事件的重点信息，分享我的感受和思考，希望能带给你启发。一次寻常的发布如往常一样，又来到了一个发布窗口，这次发生变更的迭代很简单，是支持全链路压测的一个功能上线。我们团队负责的是一个底层核心系统，链路上会有上百个应用依赖，为了应对大促这种超高流量的场景，大促前有一轮又一轮的压测。在首轮压测时，便发现我们的系统上有个数据库表不支持压测，导致压测计划无法进行。因此团队有位同事 A 就起了紧急迭代，针对业务依赖的这个数据库表做压测改造，代码变更也就几行。与此同时，同事 B 在这个系统上也想改下代码，就搭了压测改造的车，两块变更一起发布。同事 A 负责走发布流程，我们的系统有几百台服务器，部署会分为好几组，通常会搞到很晚。那天晚上，我也和大家一样，回去的比较晚，而且还忘带了手机充电器。回去没多...

2020-11-24

551

常见的数据结构链表 LinkedHashSet LinkedList 底层数据结构由链表和哈希表组成。数据的添加和删除都较为方便，就是访问比较耗费时间。数组 ArrayList 访问数据十分简单，而添加和删除数据比较耗工夫堆堆是一种图的树形结构，被用于实现“优先队列",优先队列是一种数据结构，可以自由添加数据，但取出数据时要从最小值开始按顺序取出堆的特点： ①堆中的每个结点最多有两个子结点 ②子结点必定大于父结点 ③把新数据放在最下面一行靠左的位置。当最下面一行里没有多余空间时，就再往下另起一行，把数据加在这一行的最左端 ④如果子结点的数字小于父结点的，就将父结点与其左右两个子结点中较小的一个进行交换堆中最顶端的数据始终最小，所以无论数据量有多少，取出最小值的时间复杂度都为O（1）可知树的高度为log2n，那么重构树的时间复杂度便为O（logn）栈（LIFO）略队列 (FIFO) 略哈希表 HashSet TreeSet底层数据结构是红黑树哈希函数（Hash）计算key,哈希值除以数组的长度5，求得其余数。这个余数就是key的编号即位置如果发生哈希冲突，就...

2020-11-23

474

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。