Flink中的Time與Window有什么作用

前言

Flink 是流式的、實(shí)時(shí)的 計(jì)算引擎

上面一句話就有兩個(gè)概念，一個(gè)是流式，一個(gè)是實(shí)時(shí)。

流式：就是數(shù)據(jù)源源不斷的流進(jìn)來(lái)，也就是數(shù)據(jù)沒(méi)有邊界，但是我們計(jì)算的時(shí)候必須在一個(gè)有邊界的范圍內(nèi)進(jìn)行，所以這里面就有一個(gè)問(wèn)題，邊界怎么確定？無(wú)非就兩種方式，根據(jù)時(shí)間段或者數(shù)據(jù)量進(jìn)行確定，根據(jù)時(shí)間段就是每隔多長(zhǎng)時(shí)間就劃分一個(gè)邊界，根據(jù)數(shù)據(jù)量就是每來(lái)多少條數(shù)據(jù)劃分一個(gè)邊界，F(xiàn)link 中就是這么劃分邊界的，本文會(huì)詳細(xì)講解。

實(shí)時(shí)：就是數(shù)據(jù)發(fā)送過(guò)來(lái)之后立馬就進(jìn)行相關(guān)的計(jì)算，然后將結(jié)果輸出。這里的計(jì)算有兩種：

• 一種是只有邊界內(nèi)的數(shù)據(jù)進(jìn)行計(jì)算，這種好理解，比如統(tǒng)計(jì)每個(gè)用戶最近五分鐘內(nèi)瀏覽的新聞數(shù)量，就可以取最近五分鐘內(nèi)的所有數(shù)據(jù)，然后根據(jù)每個(gè)用戶分組，統(tǒng)計(jì)新聞的總數(shù)。

• 另一種是邊界內(nèi)數(shù)據(jù)與外部數(shù)據(jù)進(jìn)行關(guān)聯(lián)計(jì)算，比如：統(tǒng)計(jì)最近五分鐘內(nèi)瀏覽新聞的用戶都是來(lái)自哪些地區(qū)，這種就需要將五分鐘內(nèi)瀏覽新聞的用戶信息與 hive 中的地區(qū)維表進(jìn)行關(guān)聯(lián)，然后在進(jìn)行相關(guān)計(jì)算。

Time與Window 

Time

在Flink中，如果以時(shí)間段劃分邊界的話，那么時(shí)間就是一個(gè)極其重要的字段。

Flink中的時(shí)間有三種類(lèi)型，如下圖所示：

 

• Event Time：是事件創(chuàng)建的時(shí)間。它通常由事件中的時(shí)間戳描述，例如采集的日志數(shù)據(jù)中，每一條日志都會(huì)記錄自己的生成時(shí)間，F(xiàn)link通過(guò)時(shí)間戳分配器訪問(wèn)事件時(shí)間戳。

• Ingestion Time：是數(shù)據(jù)進(jìn)入Flink的時(shí)間。

• Processing Time：是每一個(gè)執(zhí)行基于時(shí)間操作的算子的本地系統(tǒng)時(shí)間，與機(jī)器相關(guān)，默認(rèn)的時(shí)間屬性就是Processing Time。

例如，一條日志進(jìn)入Flink的時(shí)間為2021-01-22 10:00:00.123，到達(dá)Window的系統(tǒng)時(shí)間為2021-01-22 10:00:01.234，日志的內(nèi)容如下： 
2021-01-06 18:37:15.624 INFO Fail over to rm2

對(duì)于業(yè)務(wù)來(lái)說(shuō)，要統(tǒng)計(jì)1min內(nèi)的故障日志個(gè)數(shù)，哪個(gè)時(shí)間是最有意義的？—— eventTime，因?yàn)槲覀円鶕?jù)日志的生成時(shí)間進(jìn)行統(tǒng)計(jì)。 

Window

Window，即窗口，我們前面一直提到的邊界就是這里的Window(窗口)。

官方解釋?zhuān)?strong>流式計(jì)算是一種被設(shè)計(jì)用于處理無(wú)限數(shù)據(jù)集的數(shù)據(jù)處理引擎，而無(wú)限數(shù)據(jù)集是指一種不斷增長(zhǎng)的本質(zhì)上無(wú)限的數(shù)據(jù)集，而window是一種切割無(wú)限數(shù)據(jù)為有限塊進(jìn)行處理的手段。

所以Window是無(wú)限數(shù)據(jù)流處理的核心，Window將一個(gè)無(wú)限的stream拆分成有限大小的”buckets”桶，我們可以在這些桶上做計(jì)算操作。 

Window類(lèi)型

本文剛開(kāi)始提到，劃分窗口就兩種方式：

1. 根據(jù)時(shí)間進(jìn)行截取(time-driven-window)，比如每1分鐘統(tǒng)計(jì)一次或每10分鐘統(tǒng)計(jì)一次。

2. 根據(jù)數(shù)據(jù)進(jìn)行截取(data-driven-window)，比如每5個(gè)數(shù)據(jù)統(tǒng)計(jì)一次或每50個(gè)數(shù)據(jù)統(tǒng)計(jì)一次。

 

對(duì)于TimeWindow(根據(jù)時(shí)間劃分窗口)， 可以根據(jù)窗口實(shí)現(xiàn)原理的不同分成三類(lèi)：滾動(dòng)窗口（Tumbling Window）、滑動(dòng)窗口（Sliding Window）和會(huì)話窗口（Session Window）。

1. 滾動(dòng)窗口（Tumbling Windows）

將數(shù)據(jù)依據(jù)固定的窗口長(zhǎng)度對(duì)數(shù)據(jù)進(jìn)行切片。

特點(diǎn)：時(shí)間對(duì)齊，窗口長(zhǎng)度固定，沒(méi)有重疊。

滾動(dòng)窗口分配器將每個(gè)元素分配到一個(gè)指定窗口大小的窗口中，滾動(dòng)窗口有一個(gè)固定的大小，并且不會(huì)出現(xiàn)重疊。

例如：如果你指定了一個(gè)5分鐘大小的滾動(dòng)窗口，窗口的創(chuàng)建如下圖所示：

 

適用場(chǎng)景：適合做BI統(tǒng)計(jì)等（做每個(gè)時(shí)間段的聚合計(jì)算）。

2. 滑動(dòng)窗口（Sliding Windows）

滑動(dòng)窗口是固定窗口的更廣義的一種形式，滑動(dòng)窗口由固定的窗口長(zhǎng)度和滑動(dòng)間隔組成。

特點(diǎn)：時(shí)間對(duì)齊，窗口長(zhǎng)度固定，有重疊。

滑動(dòng)窗口分配器將元素分配到固定長(zhǎng)度的窗口中，與滾動(dòng)窗口類(lèi)似，窗口的大小由窗口大小參數(shù)來(lái)配置，另一個(gè)窗口滑動(dòng)參數(shù)控制滑動(dòng)窗口開(kāi)始的頻率。因此，滑動(dòng)窗口如果滑動(dòng)參數(shù)小于窗口大小的話，窗口是可以重疊的，在這種情況下元素會(huì)被分配到多個(gè)窗口中。

例如，你有10分鐘的窗口和5分鐘的滑動(dòng)，那么每個(gè)窗口中5分鐘的窗口里包含著上個(gè)10分鐘產(chǎn)生的數(shù)據(jù)，如下圖所示：

 

適用場(chǎng)景：對(duì)最近一個(gè)時(shí)間段內(nèi)的統(tǒng)計(jì)（求某接口最近5min的失敗率來(lái)決定是否要報(bào)警）。

3. 會(huì)話窗口（Session Windows）

由一系列事件組合一個(gè)指定時(shí)間長(zhǎng)度的timeout間隙組成，類(lèi)似于web應(yīng)用的session，也就是一段時(shí)間沒(méi)有接收到新數(shù)據(jù)就會(huì)生成新的窗口。

特點(diǎn)：時(shí)間無(wú)對(duì)齊。

session窗口分配器通過(guò)session活動(dòng)來(lái)對(duì)元素進(jìn)行分組，session窗口跟滾動(dòng)窗口和滑動(dòng)窗口相比，不會(huì)有重疊和固定的開(kāi)始時(shí)間和結(jié)束時(shí)間的情況，相反，當(dāng)它在一個(gè)固定的時(shí)間周期內(nèi)不再收到元素，即非活動(dòng)間隔產(chǎn)生，那個(gè)這個(gè)窗口就會(huì)關(guān)閉。一個(gè)session窗口通過(guò)一個(gè)session間隔來(lái)配置，這個(gè)session間隔定義了非活躍周期的長(zhǎng)度，當(dāng)這個(gè)非活躍周期產(chǎn)生，那么當(dāng)前的session將關(guān)閉并且后續(xù)的元素將被分配到新的session窗口中去。

 

Window API 

TimeWindow

TimeWindow是將指定時(shí)間范圍內(nèi)的所有數(shù)據(jù)組成一個(gè)window，一次對(duì)一個(gè)window里面的所有數(shù)據(jù)進(jìn)行計(jì)算（就是本文開(kāi)頭說(shuō)的對(duì)一個(gè)邊界內(nèi)的數(shù)據(jù)進(jìn)行計(jì)算）。

我們以 紅綠燈路口通過(guò)的汽車(chē)數(shù)量為例子：

紅綠燈路口會(huì)有汽車(chē)通過(guò)，一共會(huì)有多少汽車(chē)通過(guò)，無(wú)法計(jì)算。因?yàn)檐?chē)流源源不斷，計(jì)算沒(méi)有邊界。

所以我們統(tǒng)計(jì)每15秒鐘通過(guò)紅路燈的汽車(chē)數(shù)量，如第一個(gè)15秒為2輛，第二個(gè)15秒為3輛，第三個(gè)15秒為1輛 …

• tumbling-time-window (無(wú)重疊數(shù)據(jù))

我們使用 Linux 中的 nc 命令模擬數(shù)據(jù)的發(fā)送方

 11.開(kāi)啟發(fā)送端口，端口號(hào)為9999
2nc -lk 9999
3
42.發(fā)送內(nèi)容（key 代表不同的路口，value 代表每次通過(guò)的車(chē)輛）
5一次發(fā)送一行，發(fā)送的時(shí)間間隔代表汽車(chē)經(jīng)過(guò)的時(shí)間間隔
69,3
79,2
89,7
94,9
102,6
111,5
122,3
135,7
145,4

Flink 進(jìn)行采集數(shù)據(jù)并計(jì)算：

 1object Window {
2  def main(args: Array[String]): Unit = {
3    //TODO time-window
4    //1.創(chuàng)建運(yùn)行環(huán)境
5    val env = StreamExecutionEnvironment.getExecutionEnvironment
6
7    //2.定義數(shù)據(jù)流來(lái)源
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)計(jì)操作，每個(gè)sensorId一個(gè)tumbling窗口，窗口的大小為5秒
20    //也就是說(shuō)，每5秒鐘統(tǒng)計(jì)一次，在這過(guò)去的5秒鐘內(nèi)，各個(gè)路口通過(guò)紅綠燈汽車(chē)的數(shù)量。
21    val ds2: DataStream[CarWc] = ds1
22      .keyBy("sensorId")
23      .timeWindow(Time.seconds(5))
24      .sum("carCnt")
25
26    //5.顯示統(tǒng)計(jì)結(jié)果
27    ds2.print()
28
29    //6.觸發(fā)流計(jì)算
30    env.execute(this.getClass.getName)
31
32  }
33}

我們發(fā)送的數(shù)據(jù)并沒(méi)有指定時(shí)間字段，所以Flink使用的是默認(rèn)的 Processing Time，也就是Flink系統(tǒng)處理數(shù)據(jù)時(shí)的時(shí)間。

• sliding-time-window (有重疊數(shù)據(jù))

 1//1.創(chuàng)建運(yùn)行環(huán)境
2val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4//2.定義數(shù)據(jù)流來(lái)源
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)計(jì)操作，每個(gè)sensorId一個(gè)sliding窗口，窗口時(shí)間10秒,滑動(dòng)時(shí)間5秒
17//也就是說(shuō)，每5秒鐘統(tǒng)計(jì)一次，在這過(guò)去的10秒鐘內(nèi)，各個(gè)路口通過(guò)紅綠燈汽車(chē)的數(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)計(jì)結(jié)果
24ds2.print()
25
26//6.觸發(fā)流計(jì)算
27env.execute(this.getClass.getName)  
  

CountWindow

CountWindow根據(jù)窗口中相同key元素的數(shù)量來(lái)觸發(fā)執(zhí)行，執(zhí)行時(shí)只計(jì)算元素?cái)?shù)量達(dá)到窗口大小的key對(duì)應(yīng)的結(jié)果。

注意：CountWindow的window_size指的是相同Key的元素的個(gè)數(shù)，不是輸入的所有元素的總數(shù)。

• tumbling-count-window (無(wú)重疊數(shù)據(jù))

 1//1.創(chuàng)建運(yùn)行環(huán)境
2val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4//2.定義數(shù)據(jù)流來(lái)源
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)計(jì)操作，每個(gè)sensorId一個(gè)tumbling窗口，窗口的大小為5
17//按照key進(jìn)行收集，對(duì)應(yīng)的key出現(xiàn)的次數(shù)達(dá)到5次作為一個(gè)結(jié)果
18val ds2: DataStream[CarWc] = ds1
19  .keyBy("sensorId")
20  .countWindow(5)
21  .sum("carCnt")
22
23//5.顯示統(tǒng)計(jì)結(jié)果
24ds2.print()
25
26//6.觸發(fā)流計(jì)算
27env.execute(this.getClass.getName)

• sliding-count-window (有重疊數(shù)據(jù))

同樣也是窗口長(zhǎng)度和滑動(dòng)窗口的操作：窗口長(zhǎng)度是5，滑動(dòng)長(zhǎng)度是3

 1//1.創(chuàng)建運(yùn)行環(huán)境
2val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4//2.定義數(shù)據(jù)流來(lái)源
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)計(jì)操作，每個(gè)sensorId一個(gè)sliding窗口，窗口大小3條數(shù)據(jù),窗口滑動(dòng)為3條數(shù)據(jù)
17//也就是說(shuō)，每個(gè)路口分別統(tǒng)計(jì)，收到關(guān)于它的3條消息時(shí)統(tǒng)計(jì)在最近5條消息中，各自路口通過(guò)的汽車(chē)數(shù)量
18val ds2: DataStream[CarWc] = ds1
19  .keyBy("sensorId")
20  .countWindow(5, 3)
21  .sum("carCnt")
22
23//5.顯示統(tǒng)計(jì)結(jié)果
24ds2.print()
25
26//6.觸發(fā)流計(jì)算
27env.execute(this.getClass.getName)

• Window 總結(jié)

1. flink支持兩種劃分窗口的方式（time和count）

• 如果根據(jù)時(shí)間劃分窗口，那么它就是一個(gè)time-window

• 如果根據(jù)數(shù)據(jù)劃分窗口，那么它就是一個(gè)count-window

2. flink支持窗口的兩個(gè)重要屬性（size和interval）

• 如果size=interval,那么就會(huì)形成tumbling-window(無(wú)重疊數(shù)據(jù))

• 如果size>interval,那么就會(huì)形成sliding-window(有重疊數(shù)據(jù))

• 如果size<interval,那么這種窗口將會(huì)丟失數(shù)據(jù)。比如每5秒鐘，統(tǒng)計(jì)過(guò)去3秒的通過(guò)路口汽車(chē)的數(shù)據(jù)，將會(huì)漏掉2秒鐘的數(shù)據(jù)。

3. 通過(guò)組合可以得出四種基本窗口

• time-tumbling-window 無(wú)重疊數(shù)據(jù)的時(shí)間窗口，設(shè)置方式舉例：timeWindow(Time.seconds(5))

• time-sliding-window  有重疊數(shù)據(jù)的時(shí)間窗口，設(shè)置方式舉例：timeWindow(Time.seconds(5), Time.seconds(3))

• count-tumbling-window無(wú)重疊數(shù)據(jù)的數(shù)量窗口，設(shè)置方式舉例：countWindow(5)

• count-sliding-window 有重疊數(shù)據(jù)的數(shù)量窗口，設(shè)置方式舉例：countWindow(5,3) 

Window Reduce

WindowedStream → DataStream：給window賦一個(gè)reduce功能的函數(shù)，并返回一個(gè)聚合的結(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    // 對(duì)stream進(jìn)行處理并按key聚合
14    val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)
15
16    // 引入時(shí)間窗口
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ù)寫(xiě)入文件
25    streamReduce.print()
26
27    // 執(zhí)行程序
28    env.execute("TumblingWindow")
29  }
30}  
  

Window Apply

apply方法可以進(jìn)行一些自定義處理，通過(guò)匿名內(nèi)部類(lèi)的方法來(lái)實(shí)現(xiàn)。當(dāng)有一些復(fù)雜計(jì)算時(shí)使用。

用法

1. 實(shí)現(xiàn)一個(gè) WindowFunction 類(lèi)

2. 指定該類(lèi)的泛型為 [輸入數(shù)據(jù)類(lèi)型, 輸出數(shù)據(jù)類(lèi)型, keyBy中使用分組字段的類(lèi)型, 窗口類(lèi)型]

示例：使用apply方法來(lái)實(shí)現(xiàn)單詞統(tǒng)計(jì)

步驟：

1. 獲取流處理運(yùn)行環(huán)境

2. 構(gòu)建socket流數(shù)據(jù)源，并指定IP地址和端口號(hào)

3. 對(duì)接收到的數(shù)據(jù)轉(zhuǎn)換成單詞元組

4. 使用 keyBy 進(jìn)行分流（分組）

5. 使用 timeWinodw 指定窗口的長(zhǎng)度（每3秒計(jì)算一次）

6. 實(shí)現(xiàn)一個(gè)WindowFunction匿名內(nèi)部類(lèi)

• apply方法中實(shí)現(xiàn)聚合計(jì)算

• 使用Collector.collect收集數(shù)據(jù)

核心代碼如下：

 1    //1. 獲取流處理運(yùn)行環(huán)境
2    val env = StreamExecutionEnvironment.getExecutionEnvironment
3
4    //2. 構(gòu)建socket流數(shù)據(jù)源，并指定IP地址和端口號(hào)
5    val textDataStream = env.socketTextStream("node01", 9999).flatMap(_.split(" "))
6
7    //3. 對(duì)接收到的數(shù)據(jù)轉(zhuǎn)換成單詞元組
8    val wordDataStream = textDataStream.map(_->1)
9
10    //4. 使用 keyBy 進(jìn)行分流（分組）
11    val groupedDataStream: KeyedStream[(String, Int), String] = wordDataStream.keyBy(_._1)
12
13    //5. 使用 timeWinodw 指定窗口的長(zhǎng)度（每3秒計(jì)算一次）
14    val windowDataStream: WindowedStream[(String, Int), String, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(3))
15
16    //6. 實(shí)現(xiàn)一個(gè)WindowFunction匿名內(nèi)部類(lèi)
17    val reduceDatStream: DataStream[(String, Int)] = windowDataStream.apply(new RichWindowFunction[(String, Int), (String, Int), String, TimeWindow] {
18      //在apply方法中實(shí)現(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ù)收集起來(lái)，發(fā)送回去
25        out.collect(tuple)
26      }
27    })
28    reduceDatStream.print()
29    env.execute()  
  

Window Fold

WindowedStream → DataStream：給窗口賦一個(gè)fold功能的函數(shù)，并返回一個(gè)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    // 對(duì)stream進(jìn)行處理并按key聚合
14    val streamKeyBy = stream.map(item => (item, 1)).keyBy(0)
15
16    // 引入滾動(dòng)窗口
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ù)寫(xiě)入文件
26    streamFold.print()
27
28    // 執(zhí)行程序
29    env.execute("TumblingWindow")
30  }
31}  
  

Aggregation on Window

WindowedStream → DataStream：對(duì)一個(gè)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    // 對(duì)stream進(jìn)行處理并按key聚合
14    val streamKeyBy = stream.map(item => (item.split(" ")(0), item.split(" ")(1))).keyBy(0)
15
16    // 引入滾動(dòng)窗口
17    val streamWindow = streamKeyBy.timeWindow(Time.seconds(5))
18
19    // 執(zhí)行聚合操作
20    val streamMax = streamWindow.max(1)
21
22    // 將聚合數(shù)據(jù)寫(xiě)入文件
23    streamMax.print()
24
25    // 執(zhí)行程序
26    env.execute("TumblingWindow")
27  }
28}  
  

EventTime與Window 

EventTime的引入

1. 與現(xiàn)實(shí)世界中的時(shí)間是不一致的，在flink中被劃分為事件時(shí)間，提取時(shí)間，處理時(shí)間三種。

2. 如果以EventTime為基準(zhǔn)來(lái)定義時(shí)間窗口那將形成EventTimeWindow,要求消息本身就應(yīng)該攜帶EventTime

3. 如果以IngesingtTime為基準(zhǔn)來(lái)定義時(shí)間窗口那將形成IngestingTimeWindow,以source的systemTime為準(zhǔn)。

4. 如果以ProcessingTime基準(zhǔn)來(lái)定義時(shí)間窗口那將形成ProcessingTimeWindow，以operator的systemTime為準(zhǔn)。

在Flink的流式處理中，絕大部分的業(yè)務(wù)都會(huì)使用eventTime，一般只在eventTime無(wú)法使用時(shí)，才會(huì)被迫使用ProcessingTime或者IngestionTime。

如果要使用EventTime，那么需要引入EventTime的時(shí)間屬性，引入方式如下所示：

1val env = StreamExecutionEnvironment.getExecutionEnvironment
2
3// 從調(diào)用時(shí)刻開(kāi)始給env創(chuàng)建的每一個(gè)stream追加時(shí)間特征
4env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)  
  

Watermark 

引入

我們知道，流處理從事件產(chǎn)生，到流經(jīng) source，再到 operator，中間是有一個(gè)過(guò)程和時(shí)間的，雖然大部分情況下，流到 operator 的數(shù)據(jù)都是按照事件產(chǎn)生的時(shí)間順序來(lái)的，但是也不排除由于網(wǎng)絡(luò)、背壓等原因，導(dǎo)致亂序的產(chǎn)生，所謂亂序，就是指 Flink 接收到的事件的先后順序不是嚴(yán)格按照事件的 Event Time 順序排列的，所以 Flink 最初設(shè)計(jì)的時(shí)候，就考慮到了網(wǎng)絡(luò)延遲，網(wǎng)絡(luò)亂序等問(wèn)題，所以提出了一個(gè)抽象概念：水?。╓aterMark）；

 

如上圖所示，就出現(xiàn)一個(gè)問(wèn)題，一旦出現(xiàn)亂序，如果只根據(jù) EventTime 決定 Window 的運(yùn)行，我們不能明確數(shù)據(jù)是否全部到位，但又不能無(wú)限期的等下去，此時(shí)必須要有個(gè)機(jī)制來(lái)保證一個(gè)特定的時(shí)間后，必須觸發(fā) Window 去進(jìn)行計(jì)算了，這個(gè)特別的機(jī)制，就是 Watermark。

Watermark 是用于處理亂序事件的，而正確的處理亂序事件，通常用 Watermark 機(jī)制結(jié)合 Window 來(lái)實(shí)現(xiàn)。

數(shù)據(jù)流中的 Watermark 用于表示 timestamp 小于 Watermark 的數(shù)據(jù)，都已經(jīng)到達(dá)了，因此，Window 的執(zhí)行也是由 Watermark 觸發(fā)的。

Watermark 可以理解成一個(gè)延遲觸發(fā)機(jī)制，我們可以設(shè)置 Watermark 的延時(shí)時(shí)長(zhǎng) t，每次系統(tǒng)會(huì)校驗(yàn)已經(jīng)到達(dá)的數(shù)據(jù)中最大的 maxEventTime，然后認(rèn)定 EventTime 小于 maxEventTime - t 的所有數(shù)據(jù)都已經(jīng)到達(dá)，如果有窗口的停止時(shí)間等于 maxEventTime – t，那么這個(gè)窗口被觸發(fā)執(zhí)行。

有序流的Watermarker如下圖所示：（Watermark設(shè)置為0）

 

亂序流的Watermarker如下圖所示：（Watermark設(shè)置為2）

 

當(dāng) Flink 接收到每一條數(shù)據(jù)時(shí)，都會(huì)產(chǎn)生一條 Watermark，這條 Watermark 就等于當(dāng)前所有到達(dá)數(shù)據(jù)中的 maxEventTime - 延遲時(shí)長(zhǎng)，也就是說(shuō)，Watermark 是由數(shù)據(jù)攜帶的，一旦數(shù)據(jù)攜帶的 Watermark 比當(dāng)前未觸發(fā)的窗口的停止時(shí)間要晚，那么就會(huì)觸發(fā)相應(yīng)窗口的執(zhí)行。由于 Watermark 是由數(shù)據(jù)攜帶的，因此，如果運(yùn)行過(guò)程中無(wú)法獲取新的數(shù)據(jù)，那么沒(méi)有被觸發(fā)的窗口將永遠(yuǎn)都不被觸發(fā)。

上圖中，我們?cè)O(shè)置的允許最大延遲到達(dá)時(shí)間為2s，所以時(shí)間戳為7s的事件對(duì)應(yīng)的Watermark是5s，時(shí)間戳為12s的事件的Watermark是10s，如果我們的窗口1是1s~5s，窗口2是6s~10s，那么時(shí)間戳為7s的事件到達(dá)時(shí)的Watermarker恰好觸發(fā)窗口1，時(shí)間戳為12s的事件到達(dá)時(shí)的Watermark恰好觸發(fā)窗口2。 

Flink對(duì)于遲到數(shù)據(jù)的處理

waterMark和Window機(jī)制解決了流式數(shù)據(jù)的亂序問(wèn)題，對(duì)于因?yàn)檠舆t而順序有誤的數(shù)據(jù)，可以根據(jù)eventTime進(jìn)行業(yè)務(wù)處理，于延遲的數(shù)據(jù)Flink也有自己的解決辦法，主要的辦法是給定一個(gè)允許延遲的時(shí)間，在該時(shí)間范圍內(nèi)仍可以接受處理延遲數(shù)據(jù)。

設(shè)置允許延遲的時(shí)間是通過(guò) allowedLateness(lateness: Time)設(shè)置

保存延遲數(shù)據(jù)則是通過(guò) sideOutputLateData(outputTag: OutputTag[T])保存

獲取延遲數(shù)據(jù)是通過(guò) DataStream.getSideOutput(tag: OutputTag[X])獲取

具體的用法如下： 

allowedLateness(lateness: Time)

1def allowedLateness(lateness: Time): WindowedStream[T, K, W] = {
2  javaStream.allowedLateness(lateness)
3  this
4} 

該方法傳入一個(gè)Time值，設(shè)置允許數(shù)據(jù)遲到的時(shí)間，這個(gè)時(shí)間和 WaterMark 中的時(shí)間概念不同。再來(lái)回顧一下：

WaterMark=數(shù)據(jù)的事件時(shí)間-允許亂序時(shí)間值

隨著新數(shù)據(jù)的到來(lái)，waterMark的值會(huì)更新為最新數(shù)據(jù)事件時(shí)間-允許亂序時(shí)間值，但是如果這時(shí)候來(lái)了一條歷史數(shù)據(jù)，waterMark值則不會(huì)更新?？偟膩?lái)說(shuō)，waterMark是為了能接收到盡可能多的亂序數(shù)據(jù)。

那這里的Time值，主要是為了等待遲到的數(shù)據(jù)，在一定時(shí)間范圍內(nèi)，如果屬于該窗口的數(shù)據(jù)到來(lái)，仍會(huì)進(jìn)行計(jì)算，后面會(huì)對(duì)計(jì)算方式仔細(xì)說(shuō)明

注意：該方法只針對(duì)于基于event-time的窗口，如果是基于processing-time，并且指定了非零的time值則會(huì)拋出異常。 

sideOutputLateData(outputTag: OutputTag[T])

1def sideOutputLateData(outputTag: OutputTag[T]): WindowedStream[T, K, W] = {
2  javaStream.sideOutputLateData(outputTag)
3  this
4} 

該方法是將遲來(lái)的數(shù)據(jù)保存至給定的outputTag參數(shù)，而OutputTag則是用來(lái)標(biāo)記延遲數(shù)據(jù)的一個(gè)對(duì)象。 

DataStream.getSideOutput(tag: OutputTag[X])

通過(guò)window等操作返回的DataStream調(diào)用該方法，傳入標(biāo)記延遲數(shù)據(jù)的對(duì)象來(lái)獲取延遲的數(shù)據(jù)。 

對(duì)延遲數(shù)據(jù)的理解

延遲數(shù)據(jù)是指：

在當(dāng)前窗口【假設(shè)窗口范圍為10-15】已經(jīng)計(jì)算之后，又來(lái)了一個(gè)屬于該窗口的數(shù)據(jù)【假設(shè)事件時(shí)間為13】，這時(shí)候仍會(huì)觸發(fā) Window 操作，這種數(shù)據(jù)就稱為延遲數(shù)據(jù)。

那么問(wèn)題來(lái)了，延遲時(shí)間怎么計(jì)算呢？

假設(shè)窗口范圍為10-15，延遲時(shí)間為2s，則只要 WaterMark<15+2，并且屬于該窗口，就能觸發(fā) Window 操作。而如果來(lái)了一條數(shù)據(jù)使得 WaterMark>=15+2，10-15這個(gè)窗口就不能再觸發(fā) Window 操作，即使新來(lái)的數(shù)據(jù)的 Event Time 屬于這個(gè)窗口時(shí)間內(nèi) 。 

Flink 關(guān)聯(lián) Hive 分區(qū)表

Flink 1.12 支持了 Hive 最新的分區(qū)作為時(shí)態(tài)表的功能，可以通過(guò) SQL 的方式直接關(guān)聯(lián) Hive 分區(qū)表的最新分區(qū)，并且會(huì)自動(dòng)監(jiān)聽(tīng)最新的 Hive 分區(qū)，當(dāng)監(jiān)控到新的分區(qū)后，會(huì)自動(dòng)地做維表數(shù)據(jù)的全量替換。通過(guò)這種方式，用戶無(wú)需編寫(xiě) DataStream 程序即可完成 Kafka 流實(shí)時(shí)關(guān)聯(lián)最新的 Hive 分區(qū)實(shí)現(xiàn)數(shù)據(jù)打?qū)?/strong>。

具體用法：

在 Sql Client 中注冊(cè) 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文件 

 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用來(lái)和Hive時(shí)態(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); 

 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 --時(shí)態(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