怎樣理解Flink處理函數(shù)中的KeyedProcessFunction類,很多新手對此不是很清楚,為了幫助大家解決這個難題,下面小編將為大家詳細講解,有這方面需求的人可以來學習下,希望你能有所收獲。
創(chuàng)新互聯(lián)專注為客戶提供全方位的互聯(lián)網(wǎng)綜合服務,包含不限于成都網(wǎng)站設計、成都做網(wǎng)站、天涯網(wǎng)絡推廣、小程序開發(fā)、天涯網(wǎng)絡營銷、天涯企業(yè)策劃、天涯品牌公關、搜索引擎seo、人物專訪、企業(yè)宣傳片、企業(yè)代運營等,從售前售中售后,我們都將竭誠為您服務,您的肯定,是我們最大的嘉獎;創(chuàng)新互聯(lián)為所有大學生創(chuàng)業(yè)者提供天涯建站搭建服務,24小時服務熱線:13518219792,官方網(wǎng)址:muchs.cn
今天要了解的KeyedProcessFunction,以及該類帶來的一些特性;
通過對比類圖可以確定,KeyedProcessFunction和ProcessFunction并無直接關系: KeyedProcessFunction用于處理KeyedStream的數(shù)據(jù)集合,相比ProcessFunction類,KeyedProcessFunction擁有更多特性,官方文檔如下圖紅框,狀態(tài)處理和定時器功能都是KeyedProcessFunction才有的: 介紹完畢,接下來通過實例來學習吧;
開發(fā)環(huán)境操作系統(tǒng):MacBook Pro 13寸, macOS Catalina 10.15.3
開發(fā)工具:IDEA ULTIMATE 2018.3
JDK:1.8.0_211
Maven:3.6.0
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協(xié)議 |
git倉庫地址(ssh) | git@github.com:zq2599/blog_demos.git | 該項目源碼的倉庫地址,ssh協(xié)議 |
這個git項目中有多個文件夾,本章的應用在<font color="blue">flinkstudy</font>文件夾下,如下圖紅框所示:
本次實戰(zhàn)的目標是學習KeyedProcessFunction,內(nèi)容如下:
監(jiān)聽本機9999端口,獲取字符串;
將每個字符串用空格分隔,轉(zhuǎn)成Tuple2實例,f0是分隔后的單詞,f1等于1;
上述Tuple2實例用f0字段分區(qū),得到KeyedStream;
KeyedSteam轉(zhuǎn)入自定義KeyedProcessFunction處理;
自定義KeyedProcessFunction的作用,是記錄每個單詞最新一次出現(xiàn)的時間,然后建一個十秒的定時器,十秒后如果發(fā)現(xiàn)這個單詞沒有再次出現(xiàn),就把這個單詞和它出現(xiàn)的總次數(shù)發(fā)送到下游算子;
繼續(xù)使用《Flink處理函數(shù)實戰(zhàn)之二:ProcessFunction類》一文中創(chuàng)建的工程flinkstudy;
創(chuàng)建bean類CountWithTimestamp,里面有三個字段,為了方便使用直接設為public:
package com.bolingcavalry.keyedprocessfunction; public class CountWithTimestamp { public String key; public long count; public long lastModified; }
創(chuàng)建FlatMapFunction的實現(xiàn)類Splitter,作用是將字符串分割后生成多個Tuple2實例,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<String, Tuple2<String, Integer>> { @Override public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception { if(StringUtils.isNullOrWhitespaceOnly(s)) { System.out.println("invalid line"); return; } for(String word : s.split(" ")) { collector.collect(new Tuple2<String, Integer>(word, 1)); } } }
最后是整個邏輯功能的主體:ProcessTime.java,這里面有自定義的KeyedProcessFunction子類,還有程序入口的main方法,代碼在下面列出來之后,還會對關鍵部分做介紹:
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 體驗KeyedProcessFunction類(時間類型是處理時間) */ public class ProcessTime { /** * KeyedProcessFunction的子類,作用是將每個單詞最新出現(xiàn)時間記錄到backend,并創(chuàng)建定時器, * 定時器觸發(fā)的時候,檢查這個單詞距離上次出現(xiàn)是否已經(jīng)達到10秒,如果是,就發(fā)射給下游算子 */ static class CountWithTimeoutFunction extends KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Long>> { // 自定義狀態(tài) private ValueState<CountWithTimestamp> 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<String, Integer> value, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception { // 取得當前是哪個單詞 Tuple currentKey = ctx.getCurrentKey(); // 從backend取得當前單詞的myState狀態(tài) CountWithTimestamp current = state.value(); // 如果myState還從未沒有賦值過,就在此初始化 if (current == null) { current = new CountWithTimestamp(); current.key = value.f0; } // 單詞數(shù)量加一 current.count++; // 取當前元素的時間戳,作為該單詞最后一次出現(xiàn)的時間 current.lastModified = ctx.timestamp(); // 重新保存到backend,包括該單詞出現(xiàn)的次數(shù),以及最后一次出現(xiàn)的時間 state.update(current); // 為當前單詞創(chuàng)建定時器,十秒后后觸發(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))); } /** * 定時器觸發(fā)后執(zhí)行的方法 * @param timestamp 這個時間戳代表的是該定時器的觸發(fā)時間 * @param ctx * @param out * @throws Exception */ @Override public void onTimer( long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Long>> out) throws Exception { // 取得當前單詞 Tuple currentKey = ctx.getCurrentKey(); // 取得該單詞的myState狀態(tài) CountWithTimestamp result = state.value(); // 當前元素是否已經(jīng)連續(xù)10秒未出現(xiàn)的標志 boolean isTimeout = false; // timestamp是定時器觸發(fā)時間,如果等于最后一次更新時間+10秒,就表示這十秒內(nèi)已經(jīng)收到過該單詞了, // 這種連續(xù)十秒沒有出現(xiàn)的元素,被發(fā)送到下游算子 if (timestamp == result.lastModified + 10000) { // 發(fā)送 out.collect(new Tuple2<String, Long>(result.key, result.count)); isTimeout = true; } // 打印數(shù)據(jù),用于核對是否符合預期 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); // 處理時間 env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime); // 監(jiān)聽本地9999端口,讀取字符串 DataStream<String> socketDataStream = env.socketTextStream("localhost", 9999); // 所有輸入的單詞,如果超過10秒沒有再次出現(xiàn),都可以通過CountWithTimeoutFunction得到 DataStream<Tuple2<String, Long>> timeOutWord = socketDataStream // 對收到的字符串用空格做分割,得到多個單詞 .flatMap(new Splitter()) // 設置時間戳分配器,用當前時間作為時間戳 .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() { @Override public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) { // 使用當前系統(tǒng)時間作為時間戳 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)); } }
上述代碼有幾處需要重點關注的:
通過assignTimestampsAndWatermarks設置時間戳的時候,getCurrentWatermark返回null,因為用不上watermark;
processElement方法中,state.value()可以取得當前單詞的狀態(tài),state.update(current)可以設置當前單詞的狀態(tài),這個功能的詳情請參考《深入了解ProcessFunction的狀態(tài)操作(Flink-1.10)》;
registerProcessingTimeTimer方法設置了定時器的觸發(fā)時間,注意這里的定時器是基于processTime,和官方demo中的eventTime是不同的;
定時器觸發(fā)后,onTimer方法被執(zhí)行,里面有這個定時器的全部信息,尤其是入?yún)imestamp,這是原本設置的該定時器的觸發(fā)時間;
在控制臺執(zhí)行命令nc -l 9999,這樣就可以從控制臺向本機的9999端口發(fā)送字符串了;
在IDEA上直接執(zhí)行ProcessTime類的main方法,程序運行就開始監(jiān)聽本機的9999端口了;
在前面的控制臺輸入aaa,然后回車,等待十秒后,IEDA的控制臺輸出以下信息,從結果可見符合預期:
繼續(xù)輸入aaa再回車,連續(xù)兩次,中間間隔不要超過10秒,結果如下圖,可見每一個Tuple2元素都有一個定時器,但是第二次輸入的aaa,其定時器在出發(fā)前,aaa的最新出現(xiàn)時間就被第三次輸入的操作給更新了,于是第二次輸入aaa的定時器中的對比操作發(fā)現(xiàn)此時距aaa的最近一次(即第三次)出現(xiàn)還未達到10秒,所以第二個元素不會發(fā)射到下游算子:
下游算子收到的所有超時信息會打印出來,如下圖紅框,只打印了數(shù)量等于1和3的記錄,等于2的時候因為在10秒內(nèi)再次輸入了aaa,因此沒有超時接收,不會在下游打印: 至此,KeyedProcessFunction處理函數(shù)的學習就完成了,其狀態(tài)讀寫和定時器操作都是很實用能力。
看完上述內(nèi)容是否對您有幫助呢?如果還想對相關知識有進一步的了解或閱讀更多相關文章,請關注創(chuàng)新互聯(lián)行業(yè)資訊頻道,感謝您對創(chuàng)新互聯(lián)的支持。
新聞名稱:怎樣理解Flink處理函數(shù)中的KeyedProcessFunction類
網(wǎng)頁鏈接:http://muchs.cn/article6/pgodig.html
成都網(wǎng)站建設公司_創(chuàng)新互聯(lián),為您提供用戶體驗、關鍵詞優(yōu)化、Google、域名注冊、營銷型網(wǎng)站建設、全網(wǎng)營銷推廣
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)