Flink處理函式實戰之二：ProcessFunction類

歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

內容：所有原創文章分類匯總及配套原始碼，涉及Java、Docker、Kubernetes、DevOPS等；

Flink處理函式實戰系列鏈接

1. 深入了解ProcessFunction的狀態操作(Flink-1.10)；
2. ProcessFunction；
3. KeyedProcessFunction類；
4. ProcessAllWindowFunction(視窗處理)；
5. CoProcessFunction(雙流處理)；

關于處理函式(Process Function)

如下圖，在常規的業務開發中，SQL、Table API、DataStream API比較常用，處于Low-level的Porcession相對用得較少，從本章開始，我們一起通過實戰來熟悉處理函式(Process Function)，看看這一系列的低級算子可以帶給我們哪些能力？

關于ProcessFunction類

處理函式有很多種，最基礎的應該ProcessFunction類，來看看它的類圖，可見有RichFunction的特性open、close，然后自己有兩個重要的方法processElement和onTimer：

常用特性如下所示：

1. 處理單個元素；
2. 訪問時間戳；
3. 旁路輸出；

接下來寫兩個應用體驗上述功能；

版本資訊

1. 開發環境作業系統：MacBook Pro 13寸， macOS Catalina 10.15.3
2. 開發工具：IDEA ULTIMATE 2018.3
3. JDK：1.8.0_211
4. Maven：3.6.0
5. Flink：1.9.2

原始碼下載

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

這個git專案中有多個檔案夾，本章的應用在flinkstudy檔案夾下，如下圖紅框所示：

創建工程

執行以下命令創建一個flink-1.9.2的應用工程：

mvn \
archetype:generate \
-DarchetypeGroupId=org.apache.flink \
-DarchetypeArtifactId=flink-quickstart-java \
-DarchetypeVersion=1.9.2

按提示輸入groupId：com.bolingcavalry，architectid：flinkdemo

第一個demo

第一個demo用來體驗以下兩個特性：

1. 處理單個元素；
2. 訪問時間戳；

創建Simple.java，內容如下：

package com.bolingcavalry.processfunction;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

public class Simple {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

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

        // 設定資料源，一共三個元素
        DataStream<Tuple2<String,Integer>> dataStream = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
            @Override
            public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                for(int i=1; i<4; i++) {

                    String name = "name" + i;
                    Integer value = https://www.cnblogs.com/bolingcavalry/archive/2020/11/20/i;
                    long timeStamp = System.currentTimeMillis();

                    // 將將資料和時間戳列印出來，用來驗證資料
                    System.out.println(String.format("source，%s, %d, %d\n",
                            name,
                            value,
                            timeStamp));

                    // 發射一個元素，并且戴上了時間戳
                    ctx.collectWithTimestamp(new Tuple2<String, Integer>(name, value), timeStamp);

                    // 為了讓每個元素的時間戳不一樣，每發射一次就延時10毫秒
                    Thread.sleep(10);
                }
            }

            @Override
            public void cancel() {

            }
        });


        // 過濾值為奇數的元素
        SingleOutputStreamOperator<String> mainDataStream = dataStream
                .process(new ProcessFunction<Tuple2<String, Integer>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {
                        // f1欄位為奇數的元素不會進入下一個算子
                        if(0 == value.f1 % 2) {
                            out.collect(String.format("processElement，%s, %d, %d\n",
                                    value.f0,
                                    value.f1,
                                    ctx.timestamp()));
                        }
                    }
                });

        // 列印結果，證明每個元素的timestamp確實可以在ProcessFunction中取得
        mainDataStream.print();

        env.execute("processfunction demo : simple");
    }
}

這里對上述代碼做個介紹：

1. 創建一個資料源，每個10毫秒發出一個元素，一共三個，型別是Tuple2，f0是個字串，f1是整形，每個元素都帶時間戳；
2. 資料源發出元素時，提前把元素的f0、f1、時間戳列印出來，和后面的資料核對是否一致；
3. 在后面的處理中，創建了ProcessFunction的匿名子類，里面可以處理上游發來的每個元素，并且還能取得每個元素的時間戳(這個能力很重要)，然后將f1欄位為奇數的元素過濾掉；
4. 最后將ProcessFunction處理過的資料列印出來，驗證處理結果是否符合預期；

直接執行Simple類，結果如下，可見過濾和提取時間戳都成功了：

第二個demo

第二個demo是實作旁路輸出(Side Outputs)，對于一個DataStream來說，可以通過旁路輸出將資料輸出到其他算子中去，而不影響原有的算子的處理，下面來演示旁路輸出：

創建SideOutput類：

package com.bolingcavalry.processfunction;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.util.ArrayList;
import java.util.List;

public class SideOutput {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

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

        // 定義OutputTag
        final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

        // 創建一個List，里面有兩個Tuple2元素
        List<Tuple2<String, Integer>> list = new ArrayList<>();
        list.add(new Tuple2("aaa", 1));
        list.add(new Tuple2("bbb", 2));
        list.add(new Tuple2("ccc", 3));

        //通過List創建DataStream
        DataStream<Tuple2<String, Integer>> fromCollectionDataStream = env.fromCollection(list);

        //所有元素都進入mainDataStream，f1欄位為奇數的元素進入SideOutput
        SingleOutputStreamOperator<String> mainDataStream = fromCollectionDataStream
                .process(new ProcessFunction<Tuple2<String, Integer>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Integer> value, Context ctx, Collector<String> out) throws Exception {

                        //進入主流程的下一個算子
                        out.collect("main, name : " + value.f0 + ", value : " + value.f1);

                        //f1欄位為奇數的元素進入SideOutput
                        if(1 == value.f1 % 2) {
                            ctx.output(outputTag, "side, name : " + value.f0 + ", value : " + value.f1);
                        }
                    }
                });

        // 禁止chanin，這樣可以在頁面上看清楚原始的DAG
        mainDataStream.disableChaining();

        // 取得旁路資料
        DataStream<String> sideDataStream = mainDataStream.getSideOutput(outputTag);

        mainDataStream.print();
        sideDataStream.print();

        env.execute("processfunction demo : sideoutput");
    }
}

這里對上述代碼做個介紹：

1. 資料源是個集合，型別是Tuple2，f0欄位是字串，f1欄位是整形；
2. ProcessFunction的匿名子類中，將每個元素的f0和f1拼接成字串，發給主流程算子，再將f1欄位為奇數的元素發到旁路輸出；
3. 資料源發出元素時，提前把元素的f0、f1、時間戳列印出來，和后面的資料核對是否一致；
4. 將主流程和旁路輸出的元素都列印出來，驗證處理結果是否符合預期；

執行SideOutput看結果，如下圖，main前綴的都是主流程算子，一共三條記錄，side前綴的是旁路輸出，只有f1欄位為奇數的兩條記錄，符合預期：

上面的操作都是在IDEA上執行的，還可以將flink單獨部署，再將上述工程構建成jar，提交到flink的jobmanager，可見DAG如下：

至此，處理函式中最簡單的ProcessFunction類的學習和實戰就完成了，接下來的文章我們會嘗試更多了型別的處理函式；

歡迎關注公眾號：程式員欣宸

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

標籤：其他

上一篇：「SpringBoot2.4新特性」jar自動瘦身

下一篇：架構師的作業都干些什么？！想做架構師必看！