Flink處理函數實戰之四：窗口處理

本文轉載自查看原文 2020-11-23 09:09 496

歡迎訪問我的GitHub

https://github.com/zq2599/blog_demos

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

Flink處理函數實戰系列鏈接

本篇概覽

本文是《Flink處理函數實戰》系列的第四篇，內容是學習以下兩個窗口相關的處理函數：

ProcessAllWindowFunction：處理每個窗口內的所有元素；
ProcessWindowFunction：處理指定key的每個窗口內的所有元素；

關於ProcessAllWindowFunction

ProcessAllWindowFunction和《Flink處理函數實戰之二：ProcessFunction類》中的ProcessFunction類相似，都是用來對上游過來的元素做處理，不過ProcessFunction是每個元素執行一次processElement方法，ProcessAllWindowFunction是每個窗口執行一次process方法（方法內可以遍歷該窗口內的所有元素）；
用類圖對比可以更形象的認識差別，下圖左側是ProcessFunction，右側是ProcessAllWindowFunction：

關於ProcessWindowFunction

ProcessWindowFunction和KeyedProcessFunction類似，都是處理分區的數據，不過KeyedProcessFunction是每個元素執行一次processElement方法，而ProcessWindowFunction是每個窗口執行一次process方法（方法內可以遍歷該key當前窗口內的所有元素）；
用類圖對比可以更形象的認識差別，下圖左側是KeyedProcessFunction，右側是ProcessWindowFunction：
另外還一個差異：ProcessWindowFunction.process方法的入參就有分區的key值，而KeyedProcessFunction.processElement方法的入參沒有這個參數，而是需要Context.getCurrentKey()才能取到分區的key值；

注意事項

窗口處理函數的process方法，以ProcessAllWindowFunction為例，如下圖紅框所示，其入參可以遍歷當前窗口內的所有元素，這意味着當前窗口的所有元素都保存在堆內存中，所以請在設計階段就嚴格控制窗口內元素的內存使用量，避免耗盡TaskManager節點的堆內存：
在這里插入圖片描述
接下來通過實戰學習ProcessAllWindowFunction和ProcessWindowFunction；

版本信息

開發環境操作系統：MacBook Pro 13寸， macOS Catalina 10.15.4
開發工具：IntelliJ IDEA 2019.3.2 (Ultimate Edition)
JDK：1.8.0_121
Maven：3.3.9
Flink：1.9.2

源碼下載

如果您不想寫代碼，整個系列的源碼可在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文件夾下，如下圖紅框所示：
在這里插入圖片描述

如何實戰ProcessAllWindowFunction

接下來通過以下方式驗證ProcessAllWindowFunction功能：

每隔1秒發出一個Tuple2<String, Integer>對象，對象的f0字段在aaa和bbb之間變化，f1字段固定為1；
設置5秒的滾動窗口；
自定義ProcessAllWindowFunction擴展類，功能是統計每個窗口內元素的數量，將統計結果發給下游算子；
下游算子將統計結果打印出來；
核對發出的數據和統計信息，看是否一致；

開始編碼

繼續使用《Flink處理函數實戰之二：ProcessFunction類》一文中創建的工程flinkstudy；
新建ProcessAllWindowFunctionDemo類，如下：

package com.bolingcavalry.processwindowfunction;

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.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.text.SimpleDateFormat;
import java.util.Date;

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

        // 使用事件時間
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        // 並行度為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<Integer.MAX_VALUE; i++) {
                    // 只有aaa和bbb兩種name
                    String name = 0==i%2 ? "aaa" : "bbb";

                    // 使用當前時間作為時間戳
                    long timeStamp = System.currentTimeMillis();

                    // 將數據和時間戳打印出來，用來驗證數據
                    System.out.println(String.format("source，%s, %s\n",
                            name,
                            time(timeStamp)));

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

                    // 每發射一次就延時1秒
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        // 將數據用5秒的滾動窗口做划分，再用ProcessAllWindowFunction
        SingleOutputStreamOperator<String> mainDataStream = dataStream
                // 5秒一次的滾動窗口
                .timeWindowAll(Time.seconds(5))
                // 統計當前窗口內的元素數量，然后把數量、窗口起止時間整理成字符串發送給下游算子
                .process(new ProcessAllWindowFunction<Tuple2<String, Integer>, String, TimeWindow>() {
                    @Override
                    public void process(Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<String> collector) throws Exception {
                        int count = 0;

                        // iterable可以訪問當前窗口內的所有數據，
                        // 這里簡單處理，只統計了元素數量
                        for (Tuple2<String, Integer> tuple2 : iterable) {
                            count++;
                        }

                        // 將當前窗口的起止時間和元素數量整理成字符串
                        String value = String.format("window, %s - %s, %d\n",
                                // 當前窗口的起始時間
                                time(context.window().getStart()),
                                // 當前窗口的結束時間
                                time(context.window().getEnd()),
                                // 當前key在當前窗口內元素總數
                                count);

                        // 發射到下游算子
                        collector.collect(value);
                    }
                });

        // 打印結果，通過分析打印信息，檢查ProcessWindowFunction中可以處理所有key的整個窗口的數據
        mainDataStream.print();

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

    public static String time(long timeStamp) {
        return new SimpleDateFormat("hh:mm:ss").format(new Date(timeStamp));
    }
}

關於ProcessAllWindowFunctionDemo，有幾點需要注意：

a. 滾動窗口設置用timeWindowAll方法；

b. ProcessAllWindowFunction的匿名子類的process方法中，context.window().getStart()方法可以取得當前窗口的起始時間，getEnd()方法可以取得當前窗口的結束時間；

編碼結束，執行ProcessAllWindowFunctionDemo類驗證數據，如下圖，檢查其中一個窗口的元素詳情和ProcessAllWindowFunction執行結果，可見符合預期：
ProcessAllWindowFunction已經了解，接下來嘗試ProcessWindowFunction；

如何實戰ProcessWindowFunction

接下來通過以下方式驗證ProcessWindowFunction功能：

每隔1秒發出一個Tuple2<String, Integer>對象，對象的f0字段在aaa和bbb之間變化，f1字段固定為1；
以f0字段為key進行分區；
分區后的數據進入5秒的滾動窗口；
自定義ProcessWindowFunction擴展類，功能之一是統計每個key在每個窗口內元素的數量，將統計結果發給下游算子；
功能之二是在更新當前key的元素總量，然后在狀態后端(backend)保存，這是驗證KeyedStream在處理函數中的狀態讀寫能力；
下游算子將統計結果打印出來；
核對發出的數據和統計信息（每個窗口的和總共的分別核對），看是否一致；

開始編碼

新建ProcessWindowFunctionDemo.java：

package com.bolingcavalry.processwindowfunction;


import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
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.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.text.SimpleDateFormat;
import java.util.Date;

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

        // 使用事件時間
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        // 並行度為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 {
                int aaaNum = 0;
                int bbbNum = 0;

                for(int i=1; i<Integer.MAX_VALUE; i++) {
                    // 只有aaa和bbb兩種name
                    String name = 0==i%2 ? "aaa" : "bbb";

                    //更新aaa和bbb元素的總數
                    if(0==i%2) {
                        aaaNum++;
                    } else {
                        bbbNum++;
                    }

                    // 使用當前時間作為時間戳
                    long timeStamp = System.currentTimeMillis();

                    // 將數據和時間戳打印出來，用來驗證數據
                    System.out.println(String.format("source，%s, %s,    aaa total : %d,    bbb total : %d\n",
                            name,
                            time(timeStamp),
                            aaaNum,
                            bbbNum));

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

                    // 每發射一次就延時1秒
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        // 將數據用5秒的滾動窗口做划分，再用ProcessWindowFunction
        SingleOutputStreamOperator<String> mainDataStream = dataStream
                // 以Tuple2的f0字段作為key，本例中實際上key只有aaa和bbb兩種
                .keyBy(value -> value.f0)
                // 5秒一次的滾動窗口
                .timeWindow(Time.seconds(5))
                // 統計每個key當前窗口內的元素數量，然后把key、數量、窗口起止時間整理成字符串發送給下游算子
                .process(new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {

                    // 自定義狀態
                    private ValueState<KeyCount> state;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 初始化狀態，name是myState
                        state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", KeyCount.class));
                    }

                    @Override
                    public void process(String s, Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<String> collector) throws Exception {

                        // 從backend取得當前單詞的myState狀態
                        KeyCount current = state.value();

                        // 如果myState還從未沒有賦值過，就在此初始化
                        if (current == null) {
                            current = new KeyCount();
                            current.key = s;
                            current.count = 0;
                        }

                        int count = 0;

                        // iterable可以訪問該key當前窗口內的所有數據，
                        // 這里簡單處理，只統計了元素數量
                        for (Tuple2<String, Integer> tuple2 : iterable) {
                            count++;
                        }

                        // 更新當前key的元素總數
                        current.count += count;

                        // 更新狀態到backend
                        state.update(current);

                        // 將當前key及其窗口的元素數量，還有窗口的起止時間整理成字符串
                        String value = String.format("window, %s, %s - %s, %d,    total : %d\n",
                                // 當前key
                                s,
                                // 當前窗口的起始時間
                                time(context.window().getStart()),
                                // 當前窗口的結束時間
                                time(context.window().getEnd()),
                                // 當前key在當前窗口內元素總數
                                count,
                                // 當前key出現的總數
                                current.count);

                        // 發射到下游算子
                        collector.collect(value);
                    }
                });

        // 打印結果，通過分析打印信息，檢查ProcessWindowFunction中可以處理所有key的整個窗口的數據
        mainDataStream.print();

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

    public static String time(long timeStamp) {
        return new SimpleDateFormat("hh:mm:ss").format(new Date(timeStamp));
    }

    static class KeyCount {
        /**
         * 分區key
         */
        public String key;

        /**
         * 元素總數
         */
        public long count;
    }

}

上述代碼有幾處需要關注：

a. 靜態類KeyCount.java，是用來保存每個key元素總數的數據結構；

b. timeWindow方法設置了市場為5秒的滾動窗口；

c. 每個Tuple2元素以f0為key進行分區；

d. open方法對名為myState的自定義狀態進行注冊；

e. process方法中，state.value()取得當前key的狀態，tate.update(current)更新當前key的狀態；

接下來運行ProcessWindowFunctionDemo類檢查數據，如下圖，process方法內，對窗口內元素的統計和數據源打印的一致，並且從backend取得的總數在累加后和數據源的統計信息也一致：

至此，處理函數中窗口處理相關的實戰已經完成，如果您也在學習Flink的處理函數，希望本文能給您一些參考；

你不孤單，欣宸原創一路相伴

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

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

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Flink處理函數實戰之五：CoProcessFunction(雙流處理) Flink處理函數實戰之三：KeyedProcessFunction類 Flink處理函數實戰之二：ProcessFunction類 Flink 窗口函數 Flink流處理的時間窗口 Flink的窗口處理機制（一） Flink處理函數實戰之一：深入了解ProcessFunction的狀態(Flink-1.10) flink入門到實戰（5）flink流處理從0到1 Flink 窗口聚合函數之ProcessWindowFunction實踐 Flink SQL窗口函數詳解