CoProcessFunction實戰三部曲之一：基本功能

時間 2020-11-30

標籤 java git 程序員 github shell 數據庫 apache windows api 微信欄目 Java 简体版

原文原文鏈接

歡迎訪問個人GitHub

https://github.com/zq2599/blog_demosjava

內容：全部原創文章分類彙總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；git

關於《CoProcessFunction實戰三部曲》系列

《CoProcessFunction實戰三部曲》旨在經過三次實戰，由淺入深的學習和掌握Flink低階處理函數CoProcessFunction的用法；
整個系列的開篇先介紹CoProcessFunction，而後迅速進入實戰，瞭解CoProcessFunction的基本功能；
下一篇會結合狀態，讓雙流元素的處理彼此保持關係；
終篇的實戰會加入定時器功能，確保同一個key的數據在雙流場景下可以及時處理；

版本信息

開發環境操做系統：MacBook Pro 13寸， macOS Catalina 10.15.3
開發工具：IDEA ULTIMATE 2018.3
JDK：1.8.0_211
Maven：3.6.0
Flink：1.9.2

系列文章連接

關於CoProcessFunction

CoProcessFunction的做用是同時處理兩個數據源的數據；
試想在面對兩個輸入流時，若是這兩個流的數據之間有業務關係，該如何編碼實現呢，例以下圖中的操做，同時監聽9998和9999端口，將收到的輸出分別處理後，再由同一個sink處理(打印)：
Flink支持的方式是擴展CoProcessFunction來處理，爲了更清楚認識，咱們把KeyedProcessFunction和CoProcessFunction的類圖擺在一塊兒看，以下所示：
從上圖可見，CoProcessFunction和KeyedProcessFunction的繼承關係同樣，另外CoProcessFunction自身也很簡單，在processElement1和processElement2中分別處理兩個上游流入的數據便可，而且也支持定時器設置；

本篇實戰功能簡介

本篇我們要開發的應用，其功能很是簡單，描述以下：程序員

建兩個數據源，數據分別來自本地9998和9999端口；
每一個端口收到相似aaa,123這樣的數據，轉成Tuple2實例，f0是aaa，f1是123；
在CoProcessFunction的實現類中，對每一個數據源的數據都打日誌，而後所有傳到下游算子；
下游操做是打印，所以9998和9999端口收到的全部數據都會在控制檯打印出來；
整個demo的功能以下圖所示：

接下來開始編碼；

源碼下載

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

名稱	連接	備註
項目主頁	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文件夾下，以下圖紅框所示：
shell

代碼簡介

開發一個Map算子，將字符串轉成Tuple2；
再開發抽象類AbstractCoProcessFunctionExecutor，功能包括：flink啓動、監聽端口、調用算子處理數據、雙流鏈接、將雙流處理結果打印出來；
從上面的描述可見，AbstractCoProcessFunctionExecutor作了不少事情，惟獨沒有實現雙流鏈接後的具體業務邏輯，這些沒有作的是留給子類來實現的，整個三部曲系列的重點都集中在AbstractCoProcessFunctionExecutor的子類上，把雙流鏈接後的業務邏輯作好，以下圖所示，紅色爲CoProcessFunction的業務代碼，其餘的都在抽象類中完成：

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

抽象類

抽象類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)的需求，重寫此方法便可，此方法的入參是處理過的數據集，能夠從這裏取得側輸出；

子類，對鏈接後的雙流進行操做

本篇子類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和processElement2方法的日誌代碼已經執行，而且print方法做爲最下游，將兩個數據源的數據都打印出來了，符合預期：

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

至此，我們的第一個雙流處理低階函數就完成了，對CoProcessFunction也有了最基本的認識，固然CoProcessFunction的做用遠不及此，下一篇我們藉助狀態讓processElement1和processElement2分別對方處理過的狀態，讓每一個元素的處理都和另外一個流關聯，再也不孤立；