這篇文章主要講解了“Flink中的Time與Window有什么作用”,文中的講解內(nèi)容簡單清晰,易于學(xué)習(xí)與理解,下面請大家跟著小編的思路慢慢深入,一起來研究和學(xué)習(xí)“Flink中的Time與Window有什么作用”吧!
成都創(chuàng)新互聯(lián)公司主營大竹網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司,主營網(wǎng)站建設(shè)方案,app軟件定制開發(fā),大竹h5成都小程序開發(fā)搭建,大竹網(wǎng)站營銷推廣歡迎大竹等地區(qū)企業(yè)咨詢
Flink 是流式的、實時的 計算引擎
上面一句話就有兩個概念,一個是流式,一個是實時。
流式:就是數(shù)據(jù)源源不斷的流進來,也就是數(shù)據(jù)沒有邊界,但是我們計算的時候必須在一個有邊界的范圍內(nèi)進行,所以這里面就有一個問題,邊界怎么確定?無非就兩種方式,根據(jù)時間段或者數(shù)據(jù)量進行確定,根據(jù)時間段就是每隔多長時間就劃分一個邊界,根據(jù)數(shù)據(jù)量就是每來多少條數(shù)據(jù)劃分一個邊界,F(xiàn)link 中就是這么劃分邊界的,本文會詳細講解。
實時:就是數(shù)據(jù)發(fā)送過來之后立馬就進行相關(guān)的計算,然后將結(jié)果輸出。這里的計算有兩種:
一種是只有邊界內(nèi)的數(shù)據(jù)進行計算,這種好理解,比如統(tǒng)計每個用戶最近五分鐘內(nèi)瀏覽的新聞數(shù)量,就可以取最近五分鐘內(nèi)的所有數(shù)據(jù),然后根據(jù)每個用戶分組,統(tǒng)計新聞的總數(shù)。
另一種是邊界內(nèi)數(shù)據(jù)與外部數(shù)據(jù)進行關(guān)聯(lián)計算,比如:統(tǒng)計最近五分鐘內(nèi)瀏覽新聞的用戶都是來自哪些地區(qū),這種就需要將五分鐘內(nèi)瀏覽新聞的用戶信息與 hive 中的地區(qū)維表進行關(guān)聯(lián),然后在進行相關(guān)計算。
在Flink中,如果以時間段劃分邊界的話,那么時間就是一個極其重要的字段。
Flink中的時間有三種類型,如下圖所示:
Event Time:是事件創(chuàng)建的時間。它通常由事件中的時間戳描述,例如采集的日志數(shù)據(jù)中,每一條日志都會記錄自己的生成時間,F(xiàn)link通過時間戳分配器訪問事件時間戳。
Ingestion Time:是數(shù)據(jù)進入Flink的時間。
Processing Time:是每一個執(zhí)行基于時間操作的算子的本地系統(tǒng)時間,與機器相關(guān),默認的時間屬性就是Processing Time。
例如,一條日志進入Flink的時間為2021-01-22 10:00:00.123,到達Window的系統(tǒng)時間為2021-01-22 10:00:01.234,日志的內(nèi)容如下:
2021-01-06 18:37:15.624 INFO Fail over to rm2
對于業(yè)務(wù)來說,要統(tǒng)計1min內(nèi)的故障日志個數(shù),哪個時間是最有意義的?—— eventTime,因為我們要根據(jù)日志的生成時間進行統(tǒng)計。
Window,即窗口,我們前面一直提到的邊界就是這里的Window(窗口)。
官方解釋:流式計算是一種被設(shè)計用于處理無限數(shù)據(jù)集的數(shù)據(jù)處理引擎,而無限數(shù)據(jù)集是指一種不斷增長的本質(zhì)上無限的數(shù)據(jù)集,而window是一種切割無限數(shù)據(jù)為有限塊進行處理的手段。
所以Window是無限數(shù)據(jù)流處理的核心,Window將一個無限的stream拆分成有限大小的”buckets”桶,我們可以在這些桶上做計算操作。
本文剛開始提到,劃分窗口就兩種方式:
根據(jù)時間進行截取(time-driven-window),比如每1分鐘統(tǒng)計一次或每10分鐘統(tǒng)計一次。
根據(jù)數(shù)據(jù)進行截取(data-driven-window),比如每5個數(shù)據(jù)統(tǒng)計一次或每50個數(shù)據(jù)統(tǒng)計一次。
對于TimeWindow(根據(jù)時間劃分窗口), 可以根據(jù)窗口實現(xiàn)原理的不同分成三類:滾動窗口(Tumbling Window)、滑動窗口(Sliding Window)和會話窗口(Session Window)。
滾動窗口(Tumbling Windows)
將數(shù)據(jù)依據(jù)固定的窗口長度對數(shù)據(jù)進行切片。
特點:時間對齊,窗口長度固定,沒有重疊。
滾動窗口分配器將每個元素分配到一個指定窗口大小的窗口中,滾動窗口有一個固定的大小,并且不會出現(xiàn)重疊。
例如:如果你指定了一個5分鐘大小的滾動窗口,窗口的創(chuàng)建如下圖所示:
適用場景:適合做BI統(tǒng)計等(做每個時間段的聚合計算)。
滑動窗口(Sliding Windows)
滑動窗口是固定窗口的更廣義的一種形式,滑動窗口由固定的窗口長度和滑動間隔組成。
特點:時間對齊,窗口長度固定,有重疊。
滑動窗口分配器將元素分配到固定長度的窗口中,與滾動窗口類似,窗口的大小由窗口大小參數(shù)來配置,另一個窗口滑動參數(shù)控制滑動窗口開始的頻率。因此,滑動窗口如果滑動參數(shù)小于窗口大小的話,窗口是可以重疊的,在這種情況下元素會被分配到多個窗口中。
例如,你有10分鐘的窗口和5分鐘的滑動,那么每個窗口中5分鐘的窗口里包含著上個10分鐘產(chǎn)生的數(shù)據(jù),如下圖所示:
適用場景:對最近一個時間段內(nèi)的統(tǒng)計(求某接口最近5min的失敗率來決定是否要報警)。
會話窗口(Session Windows)
由一系列事件組合一個指定時間長度的timeout間隙組成,類似于web應(yīng)用的session,也就是一段時間沒有接收到新數(shù)據(jù)就會生成新的窗口。
特點:時間無對齊。
session窗口分配器通過session活動來對元素進行分組,session窗口跟滾動窗口和滑動窗口相比,不會有重疊和固定的開始時間和結(jié)束時間的情況,相反,當它在一個固定的時間周期內(nèi)不再收到元素,即非活動間隔產(chǎn)生,那個這個窗口就會關(guān)閉。一個session窗口通過一個session間隔來配置,這個session間隔定義了非活躍周期的長度,當這個非活躍周期產(chǎn)生,那么當前的session將關(guān)閉并且后續(xù)的元素將被分配到新的session窗口中去。
TimeWindow是將指定時間范圍內(nèi)的所有數(shù)據(jù)組成一個window,一次對一個window里面的所有數(shù)據(jù)進行計算(就是本文開頭說的對一個邊界內(nèi)的數(shù)據(jù)進行計算)。
我們以 紅綠燈路口通過的汽車數(shù)量為例子:
紅綠燈路口會有汽車通過,一共會有多少汽車通過,無法計算。因為車流源源不斷,計算沒有邊界。
所以我們統(tǒng)計每15秒鐘通過紅路燈的汽車數(shù)量,如第一個15秒為2輛,第二個15秒為3輛,第三個15秒為1輛 …
tumbling-time-window (無重疊數(shù)據(jù))
我們使用 Linux 中的 nc 命令模擬數(shù)據(jù)的發(fā)送方
11.開啟發(fā)送端口,端口號為9999
2nc -lk 9999
3
42.發(fā)送內(nèi)容(key 代表不同的路口,value 代表每次通過的車輛)
5一次發(fā)送一行,發(fā)送的時間間隔代表汽車經(jīng)過的時間間隔
69,3
79,2
89,7
94,9
102,6
111,5
122,3
135,7
145,4
Flink 進行采集數(shù)據(jù)并計算:
1object Window {
2 def main(args: Array[String]): Unit = {
3 //TODO time-window
4 //1.創(chuàng)建運行環(huán)境
5 val env = StreamExecutionEnvironment.getExecutionEnvironment
6
7 //2.定義數(shù)據(jù)流來源
8 val text = env.socketTextStream("localhost", 9999)
9
10 //3.轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
11 case class CarWc(sensorId: Int, carCnt: Int)
12 val ds1: DataStream[CarWc] = text.map {
13 line => {
14 val tokens = line.split(",")
15 CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
16 }
17 }
18
19 //4.執(zhí)行統(tǒng)計操作,每個sensorId一個tumbling窗口,窗口的大小為5秒
20 //也就是說,每5秒鐘統(tǒng)計一次,在這過去的5秒鐘內(nèi),各個路口通過紅綠燈汽車的數(shù)量。
21 val ds2: DataStream[CarWc] = ds1
22 .keyBy("sensorId")
23 .timeWindow(Time.seconds(5))
24 .sum("carCnt")
25
26 //5.顯示統(tǒng)計結(jié)果
27 ds2.print()
28
29 //6.觸發(fā)流計算
30 env.execute(this.getClass.getName)
31
32 }
33}
我們發(fā)送的數(shù)據(jù)并沒有指定時間字段,所以Flink使用的是默認的 Processing Time,也就是Flink系統(tǒng)處理數(shù)據(jù)時的時間。
sliding-time-window (有重疊數(shù)據(jù))
1//1.創(chuàng)建運行環(huán)境
2val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4//2.定義數(shù)據(jù)流來源
5val text = env.socketTextStream("localhost", 9999)
6
7//3.轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
8case class CarWc(sensorId: Int, carCnt: Int)
9
10val ds1: DataStream[CarWc] = text.map {
11 line => {
12 val tokens = line.split(",")
13 CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
14 }
15}
16//4.執(zhí)行統(tǒng)計操作,每個sensorId一個sliding窗口,窗口時間10秒,滑動時間5秒
17//也就是說,每5秒鐘統(tǒng)計一次,在這過去的10秒鐘內(nèi),各個路口通過紅綠燈汽車的數(shù)量。
18val ds2: DataStream[CarWc] = ds1
19 .keyBy("sensorId")
20 .timeWindow(Time.seconds(10), Time.seconds(5))
21 .sum("carCnt")
22
23//5.顯示統(tǒng)計結(jié)果
24ds2.print()
25
26//6.觸發(fā)流計算
27env.execute(this.getClass.getName)
CountWindow根據(jù)窗口中相同key元素的數(shù)量來觸發(fā)執(zhí)行,執(zhí)行時只計算元素數(shù)量達到窗口大小的key對應(yīng)的結(jié)果。
注意:CountWindow的window_size指的是相同Key的元素的個數(shù),不是輸入的所有元素的總數(shù)。
tumbling-count-window (無重疊數(shù)據(jù))
1//1.創(chuàng)建運行環(huán)境
2val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4//2.定義數(shù)據(jù)流來源
5val text = env.socketTextStream("localhost", 9999)
6
7//3.轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
8case class CarWc(sensorId: Int, carCnt: Int)
9
10val ds1: DataStream[CarWc] = text.map {
11 (f) => {
12 val tokens = f.split(",")
13 CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
14 }
15}
16//4.執(zhí)行統(tǒng)計操作,每個sensorId一個tumbling窗口,窗口的大小為5
17//按照key進行收集,對應(yīng)的key出現(xiàn)的次數(shù)達到5次作為一個結(jié)果
18val ds2: DataStream[CarWc] = ds1
19 .keyBy("sensorId")
20 .countWindow(5)
21 .sum("carCnt")
22
23//5.顯示統(tǒng)計結(jié)果
24ds2.print()
25
26//6.觸發(fā)流計算
27env.execute(this.getClass.getName)
sliding-count-window (有重疊數(shù)據(jù))
同樣也是窗口長度和滑動窗口的操作:窗口長度是5,滑動長度是3
1//1.創(chuàng)建運行環(huán)境
2val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4//2.定義數(shù)據(jù)流來源
5val text = env.socketTextStream("localhost", 9999)
6
7//3.轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
8case class CarWc(sensorId: Int, carCnt: Int)
9
10val ds1: DataStream[CarWc] = text.map {
11 (f) => {
12 val tokens = f.split(",")
13 CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt)
14 }
15}
16//4.執(zhí)行統(tǒng)計操作,每個sensorId一個sliding窗口,窗口大小3條數(shù)據(jù),窗口滑動為3條數(shù)據(jù)
17//也就是說,每個路口分別統(tǒng)計,收到關(guān)于它的3條消息時統(tǒng)計在最近5條消息中,各自路口通過的汽車數(shù)量
18val ds2: DataStream[CarWc] = ds1
19 .keyBy("sensorId")
20 .countWindow(5, 3)
21 .sum("carCnt")
22
23//5.顯示統(tǒng)計結(jié)果
24ds2.print()
25
26//6.觸發(fā)流計算
27env.execute(this.getClass.getName)
Window 總結(jié)
flink支持兩種劃分窗口的方式(time和count)
如果根據(jù)時間劃分窗口,那么它就是一個time-window
如果根據(jù)數(shù)據(jù)劃分窗口,那么它就是一個count-window
flink支持窗口的兩個重要屬性(size和interval)
如果size=interval,那么就會形成tumbling-window(無重疊數(shù)據(jù))
如果size>interval,那么就會形成sliding-window(有重疊數(shù)據(jù))
如果size<interval,那么這種窗口將會丟失數(shù)據(jù)。比如每5秒鐘,統(tǒng)計過去3秒的通過路口汽車的數(shù)據(jù),將會漏掉2秒鐘的數(shù)據(jù)。
通過組合可以得出四種基本窗口
time-tumbling-window 無重疊數(shù)據(jù)的時間窗口,設(shè)置方式舉例:timeWindow(Time.seconds(5))
time-sliding-window 有重疊數(shù)據(jù)的時間窗口,設(shè)置方式舉例:timeWindow(Time.seconds(5), Time.seconds(3))
count-tumbling-window無重疊數(shù)據(jù)的數(shù)量窗口,設(shè)置方式舉例:countWindow(5)
count-sliding-window 有重疊數(shù)據(jù)的數(shù)量窗口,設(shè)置方式舉例:countWindow(5,3)
WindowedStream → DataStream:給window賦一個reduce功能的函數(shù),并返回一個聚合的結(jié)果。
1import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
2import org.apache.flink.api.scala._
3import org.apache.flink.streaming.api.windowing.time.Time
4
5object StreamWindowReduce {
6 def main(args: Array[String]): Unit = {
7 // 獲取執(zhí)行環(huán)境
8 val env = StreamExecutionEnvironment.getExecutionEnvironment
9
10 // 創(chuàng)建SocketSource
11 val stream = env.socketTextStream("node01", 9999)
12
13 // 對stream進行處理并按key聚合
14 val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)
15
16 // 引入時間窗口
17 val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
18
19 // 執(zhí)行聚合操作
20 val streamReduce = streamWindow.reduce(
21 (item1, item2) => (item1._1, item1._2 + item2._2)
22 )
23
24 // 將聚合數(shù)據(jù)寫入文件
25 streamReduce.print()
26
27 // 執(zhí)行程序
28 env.execute("TumblingWindow")
29 }
30}
apply方法可以進行一些自定義處理,通過匿名內(nèi)部類的方法來實現(xiàn)。當有一些復(fù)雜計算時使用。
用法
實現(xiàn)一個 WindowFunction 類
指定該類的泛型為 [輸入數(shù)據(jù)類型, 輸出數(shù)據(jù)類型, keyBy中使用分組字段的類型, 窗口類型]
示例:使用apply方法來實現(xiàn)單詞統(tǒng)計
步驟:
獲取流處理運行環(huán)境
構(gòu)建socket流數(shù)據(jù)源,并指定IP地址和端口號
對接收到的數(shù)據(jù)轉(zhuǎn)換成單詞元組
使用 keyBy 進行分流(分組)
使用 timeWinodw 指定窗口的長度(每3秒計算一次)
實現(xiàn)一個WindowFunction匿名內(nèi)部類
apply方法中實現(xiàn)聚合計算
使用Collector.collect收集數(shù)據(jù)
核心代碼如下:
1 //1. 獲取流處理運行環(huán)境
2 val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4 //2. 構(gòu)建socket流數(shù)據(jù)源,并指定IP地址和端口號
5 val textDataStream = env.socketTextStream("node01", 9999).flatMap(_.split(" "))
6
7 //3. 對接收到的數(shù)據(jù)轉(zhuǎn)換成單詞元組
8 val wordDataStream = textDataStream.map(_->1)
9
10 //4. 使用 keyBy 進行分流(分組)
11 val groupedDataStream: KeyedStream[(String, Int), String] = wordDataStream.keyBy(_._1)
12
13 //5. 使用 timeWinodw 指定窗口的長度(每3秒計算一次)
14 val windowDataStream: WindowedStream[(String, Int), String, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(3))
15
16 //6. 實現(xiàn)一個WindowFunction匿名內(nèi)部類
17 val reduceDatStream: DataStream[(String, Int)] = windowDataStream.apply(new RichWindowFunction[(String, Int), (String, Int), String, TimeWindow] {
18 //在apply方法中實現(xiàn)數(shù)據(jù)的聚合
19 override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
20 println("hello world")
21 val tuple = input.reduce((t1, t2) => {
22 (t1._1, t1._2 + t2._2)
23 })
24 //將要返回的數(shù)據(jù)收集起來,發(fā)送回去
25 out.collect(tuple)
26 }
27 })
28 reduceDatStream.print()
29 env.execute()
WindowedStream → DataStream:給窗口賦一個fold功能的函數(shù),并返回一個fold后的結(jié)果。
1import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
2import org.apache.flink.api.scala._
3import org.apache.flink.streaming.api.windowing.time.Time
4
5object StreamWindowFold {
6 def main(args: Array[String]): Unit = {
7 // 獲取執(zhí)行環(huán)境
8 val env = StreamExecutionEnvironment.getExecutionEnvironment
9
10 // 創(chuàng)建SocketSource
11 val stream = env.socketTextStream("node01", 9999,'\n',3)
12
13 // 對stream進行處理并按key聚合
14 val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)
15
16 // 引入滾動窗口
17 val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
18
19 // 執(zhí)行fold操作
20 val streamFold = streamWindow.fold(100){
21 (begin, item) =>
22 begin + item._2
23 }
24
25 // 將聚合數(shù)據(jù)寫入文件
26 streamFold.print()
27
28 // 執(zhí)行程序
29 env.execute("TumblingWindow")
30 }
31}
WindowedStream → DataStream:對一個window內(nèi)的所有元素做聚合操作。min和 minBy的區(qū)別是min返回的是最小值,而minBy返回的是包含最小值字段的元素(同樣的原理適用于 max 和 maxBy)。
1import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
2import org.apache.flink.streaming.api.windowing.time.Time
3import org.apache.flink.api.scala._
4
5object StreamWindowAggregation {
6 def main(args: Array[String]): Unit = {
7 // 獲取執(zhí)行環(huán)境
8 val env = StreamExecutionEnvironment.getExecutionEnvironment
9
10 // 創(chuàng)建SocketSource
11 val stream = env.socketTextStream("node01", 9999)
12
13 // 對stream進行處理并按key聚合
14 val streamKeyBy = stream.map(item => (item.split(" ")(0), item.split(" ")(1))).keyBy(0)
15
16 // 引入滾動窗口
17 val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
18
19 // 執(zhí)行聚合操作
20 val streamMax = streamWindow.max(1)
21
22 // 將聚合數(shù)據(jù)寫入文件
23 streamMax.print()
24
25 // 執(zhí)行程序
26 env.execute("TumblingWindow")
27 }
28}
與現(xiàn)實世界中的時間是不一致的,在flink中被劃分為事件時間,提取時間,處理時間三種。
如果以EventTime為基準來定義時間窗口那將形成EventTimeWindow,要求消息本身就應(yīng)該攜帶EventTime
如果以IngesingtTime為基準來定義時間窗口那將形成IngestingTimeWindow,以source的systemTime為準。
如果以ProcessingTime基準來定義時間窗口那將形成ProcessingTimeWindow,以operator的systemTime為準。
在Flink的流式處理中,絕大部分的業(yè)務(wù)都會使用eventTime,一般只在eventTime無法使用時,才會被迫使用ProcessingTime或者IngestionTime。
如果要使用EventTime,那么需要引入EventTime的時間屬性,引入方式如下所示:
1val env = StreamExecutionEnvironment.getExecutionEnvironment
2
3// 從調(diào)用時刻開始給env創(chuàng)建的每一個stream追加時間特征
4env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
我們知道,流處理從事件產(chǎn)生,到流經(jīng) source,再到 operator,中間是有一個過程和時間的,雖然大部分情況下,流到 operator 的數(shù)據(jù)都是按照事件產(chǎn)生的時間順序來的,但是也不排除由于網(wǎng)絡(luò)、背壓等原因,導(dǎo)致亂序的產(chǎn)生,所謂亂序,就是指 Flink 接收到的事件的先后順序不是嚴格按照事件的 Event Time 順序排列的,所以 Flink 最初設(shè)計的時候,就考慮到了網(wǎng)絡(luò)延遲,網(wǎng)絡(luò)亂序等問題,所以提出了一個抽象概念:水印(WaterMark);
如上圖所示,就出現(xiàn)一個問題,一旦出現(xiàn)亂序,如果只根據(jù) EventTime 決定 Window 的運行,我們不能明確數(shù)據(jù)是否全部到位,但又不能無限期的等下去,此時必須要有個機制來保證一個特定的時間后,必須觸發(fā) Window 去進行計算了,這個特別的機制,就是 Watermark。
Watermark 是用于處理亂序事件的,而正確的處理亂序事件,通常用 Watermark 機制結(jié)合 Window 來實現(xiàn)。
數(shù)據(jù)流中的 Watermark 用于表示 timestamp 小于 Watermark 的數(shù)據(jù),都已經(jīng)到達了,因此,Window 的執(zhí)行也是由 Watermark 觸發(fā)的。
Watermark 可以理解成一個延遲觸發(fā)機制,我們可以設(shè)置 Watermark 的延時時長 t,每次系統(tǒng)會校驗已經(jīng)到達的數(shù)據(jù)中最大的 maxEventTime,然后認定 EventTime 小于 maxEventTime - t 的所有數(shù)據(jù)都已經(jīng)到達,如果有窗口的停止時間等于 maxEventTime – t,那么這個窗口被觸發(fā)執(zhí)行。
有序流的Watermarker如下圖所示:(Watermark設(shè)置為0)
亂序流的Watermarker如下圖所示:(Watermark設(shè)置為2)
當 Flink 接收到每一條數(shù)據(jù)時,都會產(chǎn)生一條 Watermark,這條 Watermark 就等于當前所有到達數(shù)據(jù)中的 maxEventTime - 延遲時長,也就是說,Watermark 是由數(shù)據(jù)攜帶的,一旦數(shù)據(jù)攜帶的 Watermark 比當前未觸發(fā)的窗口的停止時間要晚,那么就會觸發(fā)相應(yīng)窗口的執(zhí)行。由于 Watermark 是由數(shù)據(jù)攜帶的,因此,如果運行過程中無法獲取新的數(shù)據(jù),那么沒有被觸發(fā)的窗口將永遠都不被觸發(fā)。
上圖中,我們設(shè)置的允許最大延遲到達時間為2s,所以時間戳為7s的事件對應(yīng)的Watermark是5s,時間戳為12s的事件的Watermark是10s,如果我們的窗口1是1s~5s,窗口2是6s~10s,那么時間戳為7s的事件到達時的Watermarker恰好觸發(fā)窗口1,時間戳為12s的事件到達時的Watermark恰好觸發(fā)窗口2。
waterMark和Window機制解決了流式數(shù)據(jù)的亂序問題,對于因為延遲而順序有誤的數(shù)據(jù),可以根據(jù)eventTime進行業(yè)務(wù)處理,于延遲的數(shù)據(jù)Flink也有自己的解決辦法,主要的辦法是給定一個允許延遲的時間,在該時間范圍內(nèi)仍可以接受處理延遲數(shù)據(jù)。
設(shè)置允許延遲的時間是通過 allowedLateness(lateness: Time)設(shè)置
保存延遲數(shù)據(jù)則是通過 sideOutputLateData(outputTag: OutputTag[T])保存
獲取延遲數(shù)據(jù)是通過 DataStream.getSideOutput(tag: OutputTag[X])獲取
具體的用法如下:
1def allowedLateness(lateness: Time): WindowedStream[T, K, W] = {
2 javaStream.allowedLateness(lateness)
3 this
4}
該方法傳入一個Time值,設(shè)置允許數(shù)據(jù)遲到的時間,這個時間和 WaterMark 中的時間概念不同。再來回顧一下:
WaterMark=數(shù)據(jù)的事件時間-允許亂序時間值
隨著新數(shù)據(jù)的到來,waterMark的值會更新為最新數(shù)據(jù)事件時間-允許亂序時間值,但是如果這時候來了一條歷史數(shù)據(jù),waterMark值則不會更新。總的來說,waterMark是為了能接收到盡可能多的亂序數(shù)據(jù)。
那這里的Time值,主要是為了等待遲到的數(shù)據(jù),在一定時間范圍內(nèi),如果屬于該窗口的數(shù)據(jù)到來,仍會進行計算,后面會對計算方式仔細說明
注意:該方法只針對于基于event-time的窗口,如果是基于processing-time,并且指定了非零的time值則會拋出異常。
1def sideOutputLateData(outputTag: OutputTag[T]): WindowedStream[T, K, W] = {
2 javaStream.sideOutputLateData(outputTag)
3 this
4}
該方法是將遲來的數(shù)據(jù)保存至給定的outputTag參數(shù),而OutputTag則是用來標記延遲數(shù)據(jù)的一個對象。
通過window等操作返回的DataStream調(diào)用該方法,傳入標記延遲數(shù)據(jù)的對象來獲取延遲的數(shù)據(jù)。
延遲數(shù)據(jù)是指:
在當前窗口【假設(shè)窗口范圍為10-15】已經(jīng)計算之后,又來了一個屬于該窗口的數(shù)據(jù)【假設(shè)事件時間為13】,這時候仍會觸發(fā) Window 操作,這種數(shù)據(jù)就稱為延遲數(shù)據(jù)。
那么問題來了,延遲時間怎么計算呢?
假設(shè)窗口范圍為10-15,延遲時間為2s,則只要 WaterMark<15+2,并且屬于該窗口,就能觸發(fā) Window 操作。而如果來了一條數(shù)據(jù)使得 WaterMark>=15+2,10-15這個窗口就不能再觸發(fā) Window 操作,即使新來的數(shù)據(jù)的 Event Time 屬于這個窗口時間內(nèi) 。
Flink 1.12 支持了 Hive 最新的分區(qū)作為時態(tài)表的功能,可以通過 SQL 的方式直接關(guān)聯(lián) Hive 分區(qū)表的最新分區(qū),并且會自動監(jiān)聽最新的 Hive 分區(qū),當監(jiān)控到新的分區(qū)后,會自動地做維表數(shù)據(jù)的全量替換。通過這種方式,用戶無需編寫 DataStream 程序即可完成 Kafka 流實時關(guān)聯(lián)最新的 Hive 分區(qū)實現(xiàn)數(shù)據(jù)打?qū)?/strong>。
具體用法:
在 Sql Client 中注冊 HiveCatalog:
1vim conf/sql-client-defaults.yaml
2catalogs:
3 - name: hive_catalog
4 type: hive
5 hive-conf-dir: /disk0/soft/hive-conf/ #該目錄需要包hive-site.xml文件
創(chuàng)建 Kafka 表
1CREATE TABLE hive_catalog.flink_db.kfk_fact_bill_master_12 (
2 master Row<reportDate String, groupID int, shopID int, shopName String, action int, orderStatus int, orderKey String, actionTime bigint, areaName String, paidAmount double, foodAmount double, startTime String, person double, orderSubType int, checkoutTime String>,
3proctime as PROCTIME() -- PROCTIME用來和Hive時態(tài)表關(guān)聯(lián)
4) WITH (
5 'connector' = 'kafka',
6 'topic' = 'topic_name',
7 'format' = 'json',
8 'properties.bootstrap.servers' = 'host:9092',
9 'properties.group.id' = 'flinkTestGroup',
10 'scan.startup.mode' = 'timestamp',
11 'scan.startup.timestamp-millis' = '1607844694000'
12);
Flink 事實表與 Hive 最新分區(qū)數(shù)據(jù)關(guān)聯(lián)
dim_extend_shop_info 是 Hive 中已存在的表,所以我們用 table hint 動態(tài)地開啟維表參數(shù)。
1CREATE VIEW IF NOT EXISTS hive_catalog.flink_db.view_fact_bill_master as
2SELECT * FROM
3 (select t1.*, t2.group_id, t2.shop_id, t2.group_name, t2.shop_name, t2.brand_id,
4 ROW_NUMBER() OVER (PARTITION BY groupID, shopID, orderKey ORDER BY actionTime desc) rn
5 from hive_catalog.flink_db.kfk_fact_bill_master_12 t1
6 JOIN hive_catalog.flink_db.dim_extend_shop_info
7 /*+ OPTIONS('streaming-source.enable'='true',
8 'streaming-source.partition.include' = 'latest',
9 'streaming-source.monitor-interval' = '1 h',
10 'streaming-source.partition-order' = 'partition-name') */
11 FOR SYSTEM_TIME AS OF t1.proctime AS t2 --時態(tài)表
12 ON t1.groupID = t2.group_id and t1.shopID = t2.shop_id
13 where groupID in (202042)) t where t.rn = 1
參數(shù)解釋:
streaming-source.enable開啟流式讀取 Hive 數(shù)據(jù)。
streaming-source.partition.include有以下兩種值:
latest 屬性: 只讀取最新分區(qū)數(shù)據(jù)。
all: 讀取全量分區(qū)數(shù)據(jù) ,默認值為 all,表示讀所有分區(qū),latest 只能用在 temporal join 中,用于讀取最新分區(qū)作為維表,不能直接讀取最新分區(qū)數(shù)據(jù)。
streaming-source.monitor-interval監(jiān)聽新分區(qū)生成的時間、不宜過短 、最短是1 個小時,因為目前的實現(xiàn)是每個 task 都會查詢 metastore,高頻的查可能會對metastore 產(chǎn)生過大的壓力。需要注意的是,1.12.1 放開了這個限制,但仍建議按照實際業(yè)務(wù)不要配個太短的 interval。
streaming-source.partition-order分區(qū)策略,主要有以下 3 種,其中最為推薦的是 partition-name:
partition-name 使用默認分區(qū)名稱順序加載最新分區(qū)
create-time 使用分區(qū)文件創(chuàng)建時間順序
partition-time 使用分區(qū)時間順序
感謝各位的閱讀,以上就是“Flink中的Time與Window有什么作用”的內(nèi)容了,經(jīng)過本文的學(xué)習(xí)后,相信大家對Flink中的Time與Window有什么作用這一問題有了更深刻的體會,具體使用情況還需要大家實踐驗證。這里是創(chuàng)新互聯(lián),小編將為大家推送更多相關(guān)知識點的文章,歡迎關(guān)注!
網(wǎng)站欄目:Flink中的Time與Window有什么作用
標題網(wǎng)址:http://www.chinadenli.net/article14/gjscge.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供虛擬主機、網(wǎng)頁設(shè)計公司、網(wǎng)站制作、定制網(wǎng)站、網(wǎng)站維護、品牌網(wǎng)站制作
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)