怎樣理解Flink處理函數(shù)中的KeyedProcessFunction類，很多新手對此不是很清楚，為了幫助大家解決這個(gè)難題，下面小編將為大家詳細(xì)講解，有這方面需求的人可以來學(xué)習(xí)下，希望你能有所收獲。

創(chuàng)新互聯(lián)專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務(wù)，包含不限于成都網(wǎng)站設(shè)計(jì)、成都做網(wǎng)站、天涯網(wǎng)絡(luò)推廣、小程序開發(fā)、天涯網(wǎng)絡(luò)營銷、天涯企業(yè)策劃、天涯品牌公關(guān)、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運(yùn)營等，從售前售中售后，我們都將竭誠為您服務(wù)，您的肯定，是我們最大的嘉獎；創(chuàng)新互聯(lián)為所有大學(xué)生創(chuàng)業(yè)者提供天涯建站搭建服務(wù)，24小時(shí)服務(wù)熱線：13518219792，官方網(wǎng)址：www.cdcxhl.com

今天要了解的KeyedProcessFunction，以及該類帶來的一些特性；

關(guān)于KeyedProcessFunction

通過對比類圖可以確定，KeyedProcessFunction和ProcessFunction并無直接關(guān)系： KeyedProcessFunction用于處理KeyedStream的數(shù)據(jù)集合，相比ProcessFunction類，KeyedProcessFunction擁有更多特性，官方文檔如下圖紅框，狀態(tài)處理和定時(shí)器功能都是KeyedProcessFunction才有的： 介紹完畢，接下來通過實(shí)例來學(xué)習(xí)吧；

版本信息

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

源碼下載

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

這個(gè)git項(xiàng)目中有多個(gè)文件夾，本章的應(yīng)用在flinkstudy文件夾下，如下圖紅框所示：

實(shí)戰(zhàn)簡介

本次實(shí)戰(zhàn)的目標(biāo)是學(xué)習(xí)KeyedProcessFunction，內(nèi)容如下：

1. 監(jiān)聽本機(jī)9999端口，獲取字符串；

2. 將每個(gè)字符串用空格分隔，轉(zhuǎn)成Tuple2實(shí)例，f0是分隔后的單詞，f1等于1；

3. 上述Tuple2實(shí)例用f0字段分區(qū)，得到KeyedStream；

4. KeyedSteam轉(zhuǎn)入自定義KeyedProcessFunction處理；

5. 自定義KeyedProcessFunction的作用，是記錄每個(gè)單詞最新一次出現(xiàn)的時(shí)間，然后建一個(gè)十秒的定時(shí)器，十秒后如果發(fā)現(xiàn)這個(gè)單詞沒有再次出現(xiàn)，就把這個(gè)單詞和它出現(xiàn)的總次數(shù)發(fā)送到下游算子；

編碼

1. 繼續(xù)使用《Flink處理函數(shù)實(shí)戰(zhàn)之二：ProcessFunction類》一文中創(chuàng)建的工程flinkstudy；

2. 創(chuàng)建bean類CountWithTimestamp，里面有三個(gè)字段，為了方便使用直接設(shè)為public：

package com.bolingcavalry.keyedprocessfunction;

public class CountWithTimestamp {
    public String key;

    public long count;

    public long lastModified;
}

3. 創(chuàng)建FlatMapFunction的實(shí)現(xiàn)類Splitter，作用是將字符串分割后生成多個(gè)Tuple2實(shí)例，f0是分隔后的單詞，f1等于1：

package com.bolingcavalry;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.flink.util.StringUtils;

public class Splitter implements FlatMapFunction> {
    @Override
    public void flatMap(String s, Collector> collector) throws Exception {

        if(StringUtils.isNullOrWhitespaceOnly(s)) {
            System.out.println("invalid line");
            return;
        }

        for(String word : s.split(" ")) {
            collector.collect(new Tuple2(word, 1));
        }
    }
}

4. 最后是整個(gè)邏輯功能的主體：ProcessTime.java，這里面有自定義的KeyedProcessFunction子類，還有程序入口的main方法，代碼在下面列出來之后，還會對關(guān)鍵部分做介紹：

package com.bolingcavalry.keyedprocessfunction;

import com.bolingcavalry.Splitter;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
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.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;


/**
 * @author will
 * @email zq2599@gmail.com
 * @date 2020-05-17 13:43
 * @description 體驗(yàn)KeyedProcessFunction類(時(shí)間類型是處理時(shí)間)
 */
public class ProcessTime {

    /**
     * KeyedProcessFunction的子類，作用是將每個(gè)單詞最新出現(xiàn)時(shí)間記錄到backend，并創(chuàng)建定時(shí)器，
     * 定時(shí)器觸發(fā)的時(shí)候，檢查這個(gè)單詞距離上次出現(xiàn)是否已經(jīng)達(dá)到10秒，如果是，就發(fā)射給下游算子
     */
    static class CountWithTimeoutFunction extends KeyedProcessFunction, Tuple2> {

        // 自定義狀態(tài)
        private ValueState state;

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

        @Override
        public void processElement(
                Tuple2 value,
                Context ctx,
                Collector> out) throws Exception {

            // 取得當(dāng)前是哪個(gè)單詞
            Tuple currentKey = ctx.getCurrentKey();

            // 從backend取得當(dāng)前單詞的myState狀態(tài)
            CountWithTimestamp current = state.value();

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

            // 單詞數(shù)量加一
            current.count++;

            // 取當(dāng)前元素的時(shí)間戳，作為該單詞最后一次出現(xiàn)的時(shí)間
            current.lastModified = ctx.timestamp();

            // 重新保存到backend，包括該單詞出現(xiàn)的次數(shù)，以及最后一次出現(xiàn)的時(shí)間
            state.update(current);

            // 為當(dāng)前單詞創(chuàng)建定時(shí)器，十秒后后觸發(fā)
            long timer = current.lastModified + 10000;

            ctx.timerService().registerProcessingTimeTimer(timer);

            // 打印所有信息，用于核對數(shù)據(jù)正確性
            System.out.println(String.format("process, %s, %d, lastModified : %d (%s), timer : %d (%s)\n\n",
                    currentKey.getField(0),
                    current.count,
                    current.lastModified,
                    time(current.lastModified),
                    timer,
                    time(timer)));

        }

        /**
         * 定時(shí)器觸發(fā)后執(zhí)行的方法
         * @param timestamp 這個(gè)時(shí)間戳代表的是該定時(shí)器的觸發(fā)時(shí)間
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void onTimer(
                long timestamp,
                OnTimerContext ctx,
                Collector> out) throws Exception {

            // 取得當(dāng)前單詞
            Tuple currentKey = ctx.getCurrentKey();

            // 取得該單詞的myState狀態(tài)
            CountWithTimestamp result = state.value();

            // 當(dāng)前元素是否已經(jīng)連續(xù)10秒未出現(xiàn)的標(biāo)志
            boolean isTimeout = false;

            // timestamp是定時(shí)器觸發(fā)時(shí)間，如果等于最后一次更新時(shí)間+10秒，就表示這十秒內(nèi)已經(jīng)收到過該單詞了，
            // 這種連續(xù)十秒沒有出現(xiàn)的元素，被發(fā)送到下游算子
            if (timestamp == result.lastModified + 10000) {
                // 發(fā)送
                out.collect(new Tuple2(result.key, result.count));

                isTimeout = true;
            }

            // 打印數(shù)據(jù)，用于核對是否符合預(yù)期
            System.out.println(String.format("ontimer, %s, %d, lastModified : %d (%s), stamp : %d (%s), isTimeout : %s\n\n",
                    currentKey.getField(0),
                    result.count,
                    result.lastModified,
                    time(result.lastModified),
                    timestamp,
                    time(timestamp),
                    String.valueOf(isTimeout)));
        }
    }


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

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

        // 處理時(shí)間
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        // 監(jiān)聽本地9999端口，讀取字符串
        DataStream socketDataStream = env.socketTextStream("localhost", 9999);

        // 所有輸入的單詞，如果超過10秒沒有再次出現(xiàn)，都可以通過CountWithTimeoutFunction得到
        DataStream> timeOutWord = socketDataStream
                // 對收到的字符串用空格做分割，得到多個(gè)單詞
                .flatMap(new Splitter())
                // 設(shè)置時(shí)間戳分配器，用當(dāng)前時(shí)間作為時(shí)間戳
                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks>() {

                    @Override
                    public long extractTimestamp(Tuple2 element, long previousElementTimestamp) {
                        // 使用當(dāng)前系統(tǒng)時(shí)間作為時(shí)間戳
                        return System.currentTimeMillis();
                    }

                    @Override
                    public Watermark getCurrentWatermark() {
                        // 本例不需要watermark，返回null
                        return null;
                    }
                })
                // 將單詞作為key分區(qū)
                .keyBy(0)
                // 按單詞分區(qū)后的數(shù)據(jù)，交給自定義KeyedProcessFunction處理
                .process(new CountWithTimeoutFunction());

        // 所有輸入的單詞，如果超過10秒沒有再次出現(xiàn)，就在此打印出來
        timeOutWord.print();

        env.execute("ProcessFunction demo : KeyedProcessFunction");
    }

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

上述代碼有幾處需要重點(diǎn)關(guān)注的：

1. 通過assignTimestampsAndWatermarks設(shè)置時(shí)間戳的時(shí)候，getCurrentWatermark返回null，因?yàn)橛貌簧蟱atermark；

2. processElement方法中，state.value()可以取得當(dāng)前單詞的狀態(tài)，state.update(current)可以設(shè)置當(dāng)前單詞的狀態(tài)，這個(gè)功能的詳情請參考《深入了解ProcessFunction的狀態(tài)操作(Flink-1.10)》；

3. registerProcessingTimeTimer方法設(shè)置了定時(shí)器的觸發(fā)時(shí)間，注意這里的定時(shí)器是基于processTime，和官方demo中的eventTime是不同的；

4. 定時(shí)器觸發(fā)后，onTimer方法被執(zhí)行，里面有這個(gè)定時(shí)器的全部信息，尤其是入?yún)imestamp，這是原本設(shè)置的該定時(shí)器的觸發(fā)時(shí)間；

驗(yàn)證

1. 在控制臺執(zhí)行命令nc -l 9999，這樣就可以從控制臺向本機(jī)的9999端口發(fā)送字符串了；

2. 在IDEA上直接執(zhí)行ProcessTime類的main方法，程序運(yùn)行就開始監(jiān)聽本機(jī)的9999端口了；

3. 在前面的控制臺輸入aaa，然后回車，等待十秒后，IEDA的控制臺輸出以下信息，從結(jié)果可見符合預(yù)期：

4. 繼續(xù)輸入aaa再回車，連續(xù)兩次，中間間隔不要超過10秒，結(jié)果如下圖，可見每一個(gè)Tuple2元素都有一個(gè)定時(shí)器，但是第二次輸入的aaa，其定時(shí)器在出發(fā)前，aaa的最新出現(xiàn)時(shí)間就被第三次輸入的操作給更新了，于是第二次輸入aaa的定時(shí)器中的對比操作發(fā)現(xiàn)此時(shí)距aaa的最近一次(即第三次)出現(xiàn)還未達(dá)到10秒，所以第二個(gè)元素不會發(fā)射到下游算子：

5. 下游算子收到的所有超時(shí)信息會打印出來，如下圖紅框，只打印了數(shù)量等于1和3的記錄，等于2的時(shí)候因?yàn)樵?0秒內(nèi)再次輸入了aaa，因此沒有超時(shí)接收，不會在下游打印： 至此，KeyedProcessFunction處理函數(shù)的學(xué)習(xí)就完成了，其狀態(tài)讀寫和定時(shí)器操作都是很實(shí)用能力。

看完上述內(nèi)容是否對您有幫助呢？如果還想對相關(guān)知識有進(jìn)一步的了解或閱讀更多相關(guān)文章，請關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，感謝您對創(chuàng)新互聯(lián)的支持。