歡迎訪問我的GitHub
https://github.com/zq2599/blog_demos
內(nèi)容：所有原創(chuàng)文章分類匯總及配套源碼，涉及Java、Docker、Kubernetes、DevOPS等；
Flink處理函數(shù)實戰(zhàn)系列鏈接
- 深入了解ProcessFunction的狀態(tài)操作(Flink-1.10)；
- ProcessFunction；
- KeyedProcessFunction類；
- ProcessAllWindowFunction(窗口處理)；
- CoProcessFunction(雙流處理)；
本篇概覽
本文是《Flink處理函數(shù)實戰(zhàn)》系列的第三篇，上一篇《Flink處理函數(shù)實戰(zhàn)之二：ProcessFunction類》學(xué)習(xí)了最簡單的ProcessFunction類，今天要了解的KeyedProcessFunction，以及該類帶來的一些特性；
關(guān)于KeyedProcessFunction
通過對比類圖可以確定，KeyedProcessFunction和ProcessFunction并無直接關(guān)系：
在這里插入圖片描述
KeyedProcessFunction用于處理KeyedStream的數(shù)據(jù)集合，相比ProcessFunction類，KeyedProcessFunction擁有更多特性，官方文檔如下圖紅框，狀態(tài)處理和定時器功能都是KeyedProcessFunction才有的：
在這里插入圖片描述
介紹完畢，接下來通過實例來學(xué)習(xí)吧；
版本信息
- 開發(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項目中有多個文件夾，本章的應(yīng)用在<font color="blue">flinkstudy</font>文件夾下，如下圖紅框所示：
在這里插入圖片描述
實戰(zhàn)簡介
本次實戰(zhàn)的目標(biāo)是學(xué)習(xí)KeyedProcessFunction，內(nèi)容如下：
- 監(jiān)聽本機(jī)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，里面有三個字段，為了方便使用直接設(shè)為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方法，代碼在下面列出來之后，還會對關(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 體驗KeyedProcessFunction類(時間類型是處理時間)
*/
public class ProcessTime {
/**
* KeyedProcessFunction的子類，作用是將每個單詞最新出現(xiàn)時間記錄到backend，并創(chuàng)建定時器，
* 定時器觸發(fā)的時候，檢查這個單詞距離上次出現(xiàn)是否已經(jīng)達(dá)到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 {
// 取得當(dāng)前是哪個單詞
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)前元素的時間戳，作為該單詞最后一次出現(xiàn)的時間
current.lastModified = ctx.timestamp();
// 重新保存到backend，包括該單詞出現(xiàn)的次數(shù)，以及最后一次出現(xiàn)的時間
state.update(current);
// 為當(dāng)前單詞創(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 {
// 取得當(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是定時器觸發(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ù)，用于核對是否符合預(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);
// 處理時間
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())
// 設(shè)置時間戳分配器，用當(dāng)前時間作為時間戳
.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<Tuple2<String, Integer>>() {
@Override
public long extractTimestamp(Tuple2<String, Integer> element, long previousElementTimestamp) {
// 使用當(dāng)前系統(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));
}
}
上述代碼有幾處需要重點(diǎn)關(guān)注的：
- 通過assignTimestampsAndWatermarks設(shè)置時間戳的時候，getCurrentWatermark返回null，因為用不上watermark；
- processElement方法中，state.value()可以取得當(dāng)前單詞的狀態(tài)，state.update(current)可以設(shè)置當(dāng)前單詞的狀態(tài)，這個功能的詳情請參考《深入了解ProcessFunction的狀態(tài)操作(Flink-1.10)》；
- registerProcessingTimeTimer方法設(shè)置了定時器的觸發(fā)時間，注意這里的定時器是基于processTime，和官方demo中的eventTime是不同的；
- 定時器觸發(fā)后，onTimer方法被執(zhí)行，里面有這個定時器的全部信息，尤其是入?yún)imestamp，這是原本設(shè)置的該定時器的觸發(fā)時間；
驗證
- 在控制臺執(zhí)行命令nc -l 9999，這樣就可以從控制臺向本機(jī)的9999端口發(fā)送字符串了；
- 在IDEA上直接執(zhí)行ProcessTime類的main方法，程序運(yùn)行就開始監(jiān)聽本機(jī)的9999端口了；
-
在前面的控制臺輸入aaa，然后回車，等待十秒后，IEDA的控制臺輸出以下信息，從結(jié)果可見符合預(yù)期：
在這里插入圖片描述 -
繼續(xù)輸入aaa再回車，連續(xù)兩次，中間間隔不要超過10秒，結(jié)果如下圖，可見每一個Tuple2元素都有一個定時器，但是第二次輸入的aaa，其定時器在出發(fā)前，aaa的最新出現(xiàn)時間就被第三次輸入的操作給更新了，于是第二次輸入aaa的定時器中的對比操作發(fā)現(xiàn)此時距aaa的最近一次(即第三次)出現(xiàn)還未達(dá)到10秒，所以第二個元素不會發(fā)射到下游算子：
在這里插入圖片描述 -
下游算子收到的所有超時信息會打印出來，如下圖紅框，只打印了數(shù)量等于1和3的記錄，等于2的時候因為在10秒內(nèi)再次輸入了aaa，因此沒有超時接收，不會在下游打?。?/p>
在這里插入圖片描述
至此，KeyedProcessFunction處理函數(shù)的學(xué)習(xí)就完成了，其狀態(tài)讀寫和定時器操作都是很實用能力，希望本文可以給您提供參考；
你不孤單，欣宸原創(chuàng)一路相伴
歡迎關(guān)注公眾號：程序員欣宸
微信搜索「程序員欣宸」，我是欣宸，期待與您一同暢游Java世界...
https://github.com/zq2599/blog_demos
