歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

內(nèi)容：所有原創(chuàng)文章分類匯總及配套源碼抽兆，涉及Java识补、Docker、Kubernetes辫红、DevOPS等凭涂；

關(guān)于《CoProcessFunction實戰(zhàn)三部曲》系列

• 《CoProcessFunction實戰(zhàn)三部曲》旨在通過三次實戰(zhàn)祝辣，由淺入深的學(xué)習(xí)和掌握Flink低階處理函數(shù)CoProcessFunction的用法；
• 整個系列的開篇先介紹CoProcessFunction切油，然后迅速進(jìn)入實戰(zhàn)蝙斜，了解CoProcessFunction的基本功能；
• 下一篇會結(jié)合狀態(tài)澎胡，讓雙流元素的處理彼此保持關(guān)系孕荠；
• 終篇的實戰(zhàn)會加入定時器功能，確保同一個key的數(shù)據(jù)在雙流場景下能夠及時處理滤馍；

版本信息

1. 開發(fā)環(huán)境操作系統(tǒng)：MacBook Pro 13寸岛琼， macOS Catalina 10.15.3
2. 開發(fā)工具：IDEA ULTIMATE 2018.3
3. JDK：1.8.0_211
4. Maven：3.6.0
5. Flink：1.9.2

系列文章鏈接

1. 基本功能
2. 狀態(tài)處理
3. 定時器和側(cè)輸出

關(guān)于CoProcessFunction

• CoProcessFunction的作用是同時處理兩個數(shù)據(jù)源的數(shù)據(jù)；

• 試想在面對兩個輸入流時巢株，如果這兩個流的數(shù)據(jù)之間有業(yè)務(wù)關(guān)系槐瑞，該如何編碼實現(xiàn)呢，例如下圖中的操作阁苞，同時監(jiān)聽<font color="blue">9998</font>和<font color="blue">9999</font>端口困檩，將收到的輸出分別處理后，再由同一個sink處理(打印)：

  
  
  在這里插入圖片描述

• Flink支持的方式是擴展CoProcessFunction來處理那槽，為了更清楚認(rèn)識悼沿，我們把<font color="blue">KeyedProcessFunction</font>和<font color="blue">CoProcessFunction</font>的類圖擺在一起看，如下所示：

  
  
  在這里插入圖片描述
• 從上圖可見骚灸，CoProcessFunction和KeyedProcessFunction的繼承關(guān)系一樣糟趾，另外CoProcessFunction自身也很簡單，在<font color="blue">processElement1</font>和<font color="blue">processElement2</font>中分別處理兩個上游流入的數(shù)據(jù)即可甚牲，并且也支持定時器設(shè)置义郑；

本篇實戰(zhàn)功能簡介

本篇咱們要開發(fā)的應(yīng)用，其功能非常簡單丈钙，描述如下：

1. 建兩個數(shù)據(jù)源非驮，數(shù)據(jù)分別來自本地<font color="red">9998</font>和<font color="red">9999</font>端口；
2. 每個端口收到類似<font color="blue">aaa,123</font>這樣的數(shù)據(jù)雏赦，轉(zhuǎn)成Tuple2實例劫笙，f0是<font color="blue">aaa</font>，f1是<font color="blue">123</font>星岗；
3. 在CoProcessFunction的實現(xiàn)類中填大，對每個數(shù)據(jù)源的數(shù)據(jù)都打日志，然后全部傳到下游算子伍茄；
4. 下游操作是打印栋盹，因此<font color="red">9998</font>和<font color="red">9999</font>端口收到的所有數(shù)據(jù)都會在控制臺打印出來；

5. 整個demo的功能如下圖所示：

   
   
   在這里插入圖片描述

• 接下來開始編碼敷矫；

源碼下載

如果您不想寫代碼例获，整個系列的源碼可在GitHub下載到，地址和鏈接信息如下表所示(https://github.com/zq2599/blog_demos)：

這個git項目中有多個文件夾，本章的應(yīng)用在<font color="blue">flinkstudy</font>文件夾下怎茫，如下圖紅框所示：

在這里插入圖片描述

代碼簡介

1. 開發(fā)一個Map算子收壕，將字符串轉(zhuǎn)成Tuple2；
2. 再開發(fā)抽象類<font color="blue">AbstractCoProcessFunctionExecutor</font>轨蛤，功能包括：flink啟動蜜宪、監(jiān)聽端口、調(diào)用算子處理數(shù)據(jù)祥山、雙流連接圃验、將雙流處理結(jié)果打印出來；

3. 從上面的描述可見缝呕，<font color="blue">AbstractCoProcessFunctionExecutor</font>做了很多事情澳窑，唯獨沒有實現(xiàn)雙流連接后的具體業(yè)務(wù)邏輯，這些沒有做的是留給子類來實現(xiàn)的供常，整個三部曲系列的重點都集中在AbstractCoProcessFunctionExecutor的子類上摊聋，把雙流連接后的業(yè)務(wù)邏輯做好，如下圖所示栈暇，紅色為CoProcessFunction的業(yè)務(wù)代碼麻裁，其他的都在抽象類中完成：

   
   
   在這里插入圖片描述

Map算子

1. 做一個map算子，用來將字符串<font color="blue">aaa,123</font>轉(zhuǎn)成Tuple2實例源祈，f0是<font color="red">aaa</font>煎源，f1是<font color="red">123</font>；
2. 算子名為<font color="blue">WordCountMap.java</font>：

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

抽象類

• 抽象類AbstractCoProcessFunctionExecutor.java新博，源碼如下薪夕，稍后會說明幾個關(guān)鍵點：

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 串起整個邏輯的執(zhí)行類，用于體驗CoProcessFunction
 */
public abstract class AbstractCoProcessFunctionExecutor {

    /**
     * 返回CoProcessFunction的實例赫悄，這個方法留給子類實現(xiàn)
     * @return
     */
    protected abstract CoProcessFunction<
            Tuple2<String, Integer>,
            Tuple2<String, Integer>,
            Tuple2<String, Integer>> getCoProcessFunctionInstance();

    /**
     * 監(jiān)聽根據(jù)指定的端口原献，
     * 得到的數(shù)據(jù)先通過map轉(zhuǎn)為Tuple2實例，
     * 給元素加入時間戳埂淮，
     * 再按f0字段分區(qū)姑隅，
     * 將分區(qū)后的KeyedStream返回
     * @param port
     * @return
     */
    protected KeyedStream<Tuple2<String, Integer>, Tuple> buildStreamFromSocket(StreamExecutionEnvironment env, int port) {
        return env
                // 監(jiān)聽端口
                .socketTextStream("localhost", port)
                // 得到的字符串"aaa,3"轉(zhuǎn)成Tuple2實例，f0="aaa"倔撞，f1=3
                .map(new WordCountMap())
                // 將單詞作為key分區(qū)
                .keyBy(0);
    }

    /**
     * 如果子類有側(cè)輸出需要處理讲仰，請重寫此方法，會在主流程執(zhí)行完畢后被調(diào)用
     */
    protected void doSideOutput(SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream) {
    }

    /**
     * 執(zhí)行業(yè)務(wù)的方法
     * @throws Exception
     */
    public void execute() throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

        // 監(jiān)聽9998端口的輸入
        KeyedStream<Tuple2<String, Integer>, Tuple> stream1 = buildStreamFromSocket(env, 9998);

        // 監(jiān)聽9999端口的輸入
        KeyedStream<Tuple2<String, Integer>, Tuple> stream2 = buildStreamFromSocket(env, 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream = stream1
                // 兩個流連接
                .connect(stream2)
                // 執(zhí)行低階處理函數(shù)痪蝇，具體處理邏輯在子類中實現(xiàn)
                .process(getCoProcessFunctionInstance());

        // 將低階處理函數(shù)輸出的元素全部打印出來
        mainDataStream.print();

        // 側(cè)輸出相關(guān)邏輯鄙陡，子類有側(cè)輸出需求時重寫此方法
        doSideOutput(mainDataStream);

        // 執(zhí)行
        env.execute("ProcessFunction demo : CoProcessFunction");
    }
}

• 關(guān)鍵點之一：一共有兩個數(shù)據(jù)源冕房，每個源的處理邏輯都封裝到<font color="blue">buildStreamFromSocket</font>方法中；
• 關(guān)鍵點之二：<font color="blue">stream1.connect(stream2)</font>將兩個流連接起來趁矾；
• 關(guān)鍵點之三：<font color="blue">process</font>接收CoProcessFunction實例耙册，合并后的流的處理邏輯就在這里面；
• 關(guān)鍵點之四：<font color="blue">getCoProcessFunctionInstance</font>是抽象方法毫捣，返回<font color="blue">CoProcessFunction</font>實例详拙，交給子類實現(xiàn)，所以CoProcessFunction中做什么事情完全由子類決定蔓同；
• 關(guān)鍵點之五：doSideOutput方法中啥也沒做饶辙，但是在主流程代碼的末尾會被調(diào)用，如果子類有側(cè)輸出(SideOutput)的需求斑粱，重寫此方法即可弃揽，此方法的入?yún)⑹翘幚磉^的數(shù)據(jù)集，可以從這里取得側(cè)輸出珊佣；

子類蹋宦，對連接后的雙流進(jìn)行操作

1. 本篇子類<font color="blue">CollectEveryOne.java</font>如下所示，邏輯很簡單咒锻，將每個源的上游數(shù)據(jù)直接輸出到下游算子：

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

2. 上述代碼中冷冗，CoProcessFunction后面的泛型定義很長：<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> ，一共三個Tuple2惑艇，分別代表一號數(shù)據(jù)源輸入蒿辙、二號數(shù)據(jù)源輸入、下游輸出的類型滨巴；
3. 編碼完成思灌，運行起來試試；

驗證

1. 分別開啟本機的<font color="blue">9998</font>和<font color="blue">9999</font>端口恭取，我這里是MacBook泰偿，執(zhí)行<font color="blue">nc -l 9998</font>和<font color="blue">nc -l 9999</font>
2. 啟動Flink應(yīng)用，如果您和我一樣是Mac電腦蜈垮，直接運行<font color="blue">CollectEveryOne.main</font>方法即可（如果是windows電腦耗跛，我這沒試過，不過做成jar在線部署也是可以的）攒发；
3. 在監(jiān)聽9998和9999端口的控制臺分別輸入<font color="blue">aaa,111</font>和<font color="blue">bbb,222</font>
4. 以下是flink控制臺輸出的內(nèi)容调塌，可見processElement1和processElement2方法的日志代碼已經(jīng)執(zhí)行，并且print方法作為最下游惠猿，將兩個數(shù)據(jù)源的數(shù)據(jù)都打印出來了羔砾，符合預(yù)期：

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

• 至此，咱們的第一個雙流處理低階函數(shù)就完成了，對CoProcessFunction也有了最基本的認(rèn)識姜凄，當(dāng)然CoProcessFunction的作用遠(yuǎn)不及此政溃，下一篇咱們借助狀態(tài)讓<font color="blue">processElement1</font>和<font color="blue">processElement2</font>分別對方處理過的狀態(tài)，讓每個元素的處理都和另一個流關(guān)聯(lián)檀葛，不再孤立玩祟；

你不孤單腹缩，欣宸原創(chuàng)一路相伴

1. Java系列
2. Spring系列
3. Docker系列
4. kubernetes系列
5. 數(shù)據(jù)庫+中間件系列
6. DevOps系列

歡迎關(guān)注公眾號：程序員欣宸

微信搜索「程序員欣宸」屿聋，我是欣宸，期待與您一同暢游Java世界...
https://github.com/zq2599/blog_demos