歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

內容：所有原創(chuàng)文章分類匯總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；

歡迎訪問我的GitHub

這里分類和匯總了欣宸的全部原創(chuàng)(含配套源碼)：https://github.com/zq2599/blog_demos

Flink處理函數實戰(zhàn)系列鏈接

1. 深入了解ProcessFunction的狀態(tài)操作(Flink-1.10)；

2. ProcessFunction；

3. KeyedProcessFunction類；

4. ProcessAllWindowFunction(窗口處理)；

5. CoProcessFunction(雙流處理)；

本篇概覽

• 本文是《Flink處理函數實戰(zhàn)》系列的第五篇，學習內容是如何同時處理兩個數據源的數據；

• 試想在面對兩個輸入流時，如果這兩個流的數據之間有業(yè)務關系，該如何編碼實現(xiàn)呢，例如下圖中的操作，同時監(jiān)聽9998和9999端口，將收到的輸出分別處理后，再由同一個sink處理(打印)：
  

• Flink支持的方式是擴展CoProcessFunction來處理，為了更清楚認識，我們把KeyedProcessFunction和CoProcessFunction的類圖擺在一起看，如下所示：
  

• 從上圖可見，CoProcessFunction和KeyedProcessFunction的繼承關系一樣，另外CoProcessFunction自身也很簡單，在processElement1和processElement2中分別處理兩個上游流入的數據即可，并且也支持定時器設置；

編碼實戰(zhàn)

接下來咱們開發(fā)一個應用來體驗CoProcessFunction，功能非常簡單，描述如下：

1. 建兩個數據源，數據分別來自本地9998和9999端口；

2. 每個端口收到類似aaa,123這樣的數據，轉成Tuple2實例，f0是aaa，f1是123；

3. 在CoProcessFunction的實現(xiàn)類中，對每個數據源的數據都打日志，然后全部傳到下游算子；

4. 下游操作是打印，因此9998和9999端口收到的所有數據都會在控制臺打印出來；

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

• 接下來編碼實現(xiàn)上述功能；

源碼下載

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

這個git項目中有多個文件夾，本章的應用在flinkstudy文件夾下，如下圖紅框所示：

Map算子

1. 做一個map算子，用來將字符串aaa,123轉成Tuple2實例，f0是aaa，f1是123；

2. 算子名為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]));
    }
}

便于擴展的抽象類

• 開發(fā)一個抽象類，將前面圖中提到的監(jiān)聽端口、map處理、keyby處理、打印都做到這個抽象類中，但是CoProcessFunction的邏輯卻不放在這里，而是交給子類來實現(xiàn)，這樣如果我們想進一步實踐和擴展CoProcessFunction的能力，只要在子類中專注做好CoProcessFunction相關開發(fā)即可，如下圖，紅色部分交給子類實現(xiàn)，其余的都是抽象類完成的：
  

• 抽象類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 串起整個邏輯的執(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)聽根據指定的端口，
     * 得到的數據先通過map轉為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"轉成Tuple2實例，f0="aaa"，f1=3
                .map(new WordCountMap())
                // 將單詞作為key分區(qū)
                .keyBy(0);
    }

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

    /**
     * 執(zhí)行業(yè)務的方法
     * @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í)行低階處理函數，具體處理邏輯在子類中實現(xiàn)
                .process(getCoProcessFunctionInstance());

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

        // 側輸出相關邏輯，子類有側輸出需求時重寫此方法
        doSideOutput(mainDataStream);

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

• 關鍵點之一：一共有兩個數據源，每個源的處理邏輯都封裝到buildStreamFromSocket方法中；

• 關鍵點之二：stream1.connect(stream2)將兩個流連接起來；

• 關鍵點之三：process接收CoProcessFunction實例，合并后的流的處理邏輯就在這里面；

• 關鍵點之四：getCoProcessFunctionInstance是抽象方法，返回CoProcessFunction實例，交給子類實現(xiàn)，所以CoProcessFunction中做什么事情完全由子類決定；

• 關鍵點之五：doSideOutput方法中啥也沒做，但是在主流程代碼的末尾會被調用，如果子類有側輸出(SideOutput)的需求，重寫此方法即可，此方法的入參是處理過的數據集，可以從這里取得側輸出；

子類決定CoProcessFunction的功能

1. 子類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();
    }
}

1. 上述代碼中，CoProcessFunction后面的泛型定義很長：<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> ，一共三個Tuple2，分別代表一號數據源輸入、二號數據源輸入、下游輸出的類型；

驗證

1. 分別開啟本機的9998和9999端口，我這里是MacBook，執(zhí)行nc -l 9998和nc -l 9999

2. 啟動Flink應用，如果您和我一樣是Mac電腦，直接運行CollectEveryOne.main方法即可（如果是windows電腦，我這沒試過，不過做成jar在線部署也是可以的）；

3. 在監(jiān)聽9998和9999端口的控制臺分別輸入aaa,111和bbb,222

4. 以下是flink控制臺輸出的內容，可見processElement1和processElement1方法的日志代碼已經執(zhí)行，并且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的使用遠不及此，結合狀態(tài)，可以processElement1獲得更多二號流的元素信息，另外還可以結合定時器來約束兩個流協(xié)同處理的等待時間，您可以參考前面文章中的狀態(tài)和定時器來自行嘗試；