四、Flink 算子大全

Flink和Spark類似,也是一種一站式處理的框架;既可以進(jìn)行批處理(DataSet),也可以進(jìn)行實(shí)時(shí)處理(DataStream)。

所以下面將Flink的算子分為兩大類:一類是DataSet,一類是DataStream。

DataSet 批處理算子一、Source算子1． fromCollection

fromCollection:從本地集合讀取數(shù)據(jù)

例:

val env = ExecutionEnvironment．getExecutionEnvironment
val textDataSet: DataSet[String] = env．fromCollection(
 List("1,張三", "2,李四", "3,王五", "4,趙六")
)
2． readTextFile

readTextFile:從文件中讀取

val textDataSet: DataSet[String]  = env．readTextFile("/data/a．txt")
3． readTextFile:遍歷目錄

readTextFile可以對一個(gè)文件目錄內(nèi)的所有文件,包括所有子目錄中的所有文件的遍歷訪問方式

val parameters = new Configuration
// recursive．file．enumeration 開啟遞歸
parameters．setBoolean("recursive．file．enumeration", true)
val file = env．readTextFile("/data")．withParameters(parameters)
4． readTextFile:讀取壓縮文件

對于以下壓縮類型,不需要指定任何額外的inputformat方法,flink可以自動識別并且解壓。但是,壓縮文件可能不會并行讀取,可能是順序讀取的,這樣可能會影響作業(yè)的可伸縮性。

壓縮方法文件擴(kuò)展名是否可并行讀取DEFLATE．deflatenoGZip．gz ．gzipnoBzip2．bz2noXZ．xznoval file = env．readTextFile("/data/file．gz")
二、Transform轉(zhuǎn)換算子

因?yàn)門ransform算子基于Source算子操作,所以首先構(gòu)建Flink執(zhí)行環(huán)境及Source算子,后續(xù)Transform算子操作基于此:

val env = ExecutionEnvironment．getExecutionEnvironment
val textDataSet: DataSet[String] = env．fromCollection(
 List("張三,1", "李四,2", "王五,3", "張三,4")
)
1． map

將DataSet中的每一個(gè)元素轉(zhuǎn)換為另外一個(gè)元素

// 使用map將List轉(zhuǎn)換為一個(gè)Scala的樣例類
case class User(name: String, id: String)
val userDataSet: DataSet[User] = textDataSet．map {
 text =>
   val fieldArr = text．split(",")
   User(fieldArr(0), fieldArr(1))
}
userDataSet．print()
2． flatMap

將DataSet中的每一個(gè)元素轉(zhuǎn)換為0．．．n個(gè)元素。

// 使用flatMap操作,將集合中的數(shù)據(jù):
// 根據(jù)第一個(gè)元素,進(jìn)行分組
// 根據(jù)第二個(gè)元素,進(jìn)行聚合求值
val result = textDataSet．flatMap(line => line)
     ．groupBy(0) // 根據(jù)第一個(gè)元素,進(jìn)行分組
     ．sum(1) // 根據(jù)第二個(gè)元素,進(jìn)行聚合求值
     
result．print()
3． mapPartition

將一個(gè)分區(qū)中的元素轉(zhuǎn)換為另一個(gè)元素

// 使用mapPartition操作,將List轉(zhuǎn)換為一個(gè)scala的樣例類
case class User(name: String, id: String)
val result: DataSet[User] = textDataSet．mapPartition(line => {
     line．map(index => User(index．_1, index．_2))
   })
   
result．print()
4． filter

過濾出來一些符合條件的元素,返回boolean值為true的元素

val source: DataSet[String] = env．fromElements("java", "scala", "java")
val filter:DataSet[String] = source．filter(line => line．contains("java"))//過濾出帶java的數(shù)據(jù)
filter．print()
5． reduce

可以對一個(gè)dataset或者一個(gè)group來進(jìn)行聚合計(jì)算,最終聚合成一個(gè)元素

// 使用 fromElements 構(gòu)建數(shù)據(jù)源
val source = env．fromElements(("java", 1), ("scala", 1), ("java", 1))
// 使用map轉(zhuǎn)換成DataSet元組
val mapData: DataSet[(String, Int)] = source．map(line => line)
// 根據(jù)首個(gè)元素分組
val groupData = mapData．groupBy(_．_1)
// 使用reduce聚合
val reduceData = groupData．reduce((x, y) => (x．_1, x．_2 + y．_2))
// 打印測試
reduceData．print()
6． reduceGroup

將一個(gè)dataset或者一個(gè)group聚合成一個(gè)或多個(gè)元素。
reduceGroup是reduce的一種優(yōu)化方案;
它會先分組reduce,然后在做整體的reduce;這樣做的好處就是可以減少網(wǎng)絡(luò)IO

// 使用 fromElements 構(gòu)建數(shù)據(jù)源
val source: DataSet[(String, Int)] = env．fromElements(("java", 1), ("scala", 1), ("java", 1))
// 根據(jù)首個(gè)元素分組
val groupData = source．groupBy(_．_1)
// 使用reduceGroup聚合
val result: DataSet[(String, Int)] = groupData．reduceGroup {
     (in: Iterator[(String, Int)], out: Collector[(String, Int)]) =>
       val tuple = in．reduce((x, y) => (x．_1, x．_2 + y．_2))
       out．collect(tuple)
   }
// 打印測試
result．print()
7． minBy和maxBy

選擇具有最小值或最大值的元素

// 使用minBy操作,求List中每個(gè)人的最小值
// List("張三,1", "李四,2", "王五,3", "張三,4")
case class User(name: String, id: String)
// 將List轉(zhuǎn)換為一個(gè)scala的樣例類
val text: DataSet[User] = textDataSet．mapPartition(line => {
     line．map(index => User(index．_1, index．_2))
   })
   
val result = text
         ．groupBy(0) // 按照姓名分組
         ．minBy(1)   // 每個(gè)人的最小值
8． Aggregate

在數(shù)據(jù)集上進(jìn)行聚合求最值(最大值、最小值)

val data = new mutable．MutableList[(Int, String, Double)]
   data．+=((1, "yuwen", 89．0))
   data．+=((2, "shuxue", 92．2))
   data．+=((3, "yuwen", 89．99))
// 使用 fromElements 構(gòu)建數(shù)據(jù)源
val input: DataSet[(Int, String, Double)] = env．fromCollection(data)
// 使用group執(zhí)行分組操作
val value = input．groupBy(1)
           // 使用aggregate求最大值元素
           ．a(chǎn)ggregate(Aggregations．MAX, 2)
// 打印測試
value．print()      

Aggregate只能作用于元組上

注意:
要使用aggregate,只能使用字段索引名或索引名稱來進(jìn)行分組 groupBy(0) ,否則會報(bào)一下錯(cuò)誤:
Exception in thread "main" java．lang．UnsupportedOperationException: Aggregate does not supportgrouping with KeySelector functions, yet．

9． distinct

去除重復(fù)的數(shù)據(jù)

// 數(shù)據(jù)源使用上一題的
// 使用distinct操作,根據(jù)科目去除集合中重復(fù)的元組數(shù)據(jù)
val value: DataSet[(Int, String, Double)] = input．distinct(1)
value．print()
10． first

取前N個(gè)數(shù)

input．first(2) // 取前兩個(gè)數(shù)
11． join

將兩個(gè)DataSet按照一定條件連接到一起,形成新的DataSet

// s1 和 s2 數(shù)據(jù)集格式如下:
// DataSet[(Int, String,String, Double)]
val joinData = s1．join(s2)  // s1數(shù)據(jù)集 join s2數(shù)據(jù)集
            ．where(0)．equalTo(0) {     // join的條件
     (s1, s2) => (s1．_1, s1．_2, s2．_2, s1．_3)
   }
12． leftOuterJoin

左外連接,左邊的Dataset中的每一個(gè)元素,去連接右邊的元素

此外還有:

rightOuterJoin:右外連接,左邊的Dataset中的每一個(gè)元素,去連接左邊的元素

fullOuterJoin:全外連接,左右兩邊的元素,全部連接

下面以 leftOuterJoin 進(jìn)行示例:

val data1 = ListBuffer[Tuple2[Int,String]]()
   data1．a(chǎn)ppend((1,"zhangsan"))
   data1．a(chǎn)ppend((2,"lisi"))
   data1．a(chǎn)ppend((3,"wangwu"))
   data1．a(chǎn)ppend((4,"zhaoliu"))
val data2 = ListBuffer[Tuple2[Int,String]]()
   data2．a(chǎn)ppend((1,"beijing"))
   data2．a(chǎn)ppend((2,"shanghai"))
   data2．a(chǎn)ppend((4,"guangzhou"))
val text1 = env．fromCollection(data1)
val text2 = env．fromCollection(data2)
text1．leftOuterJoin(text2)．where(0)．equalTo(0)．a(chǎn)pply((first,second)=>{
     if(second==null){
       (first．_1,first．_2,"null")
     }else{
       (first．_1,first．_2,second．_2)
     }
   })．print()
13． cross

交叉操作,通過形成這個(gè)數(shù)據(jù)集和其他數(shù)據(jù)集的笛卡爾積,創(chuàng)建一個(gè)新的數(shù)據(jù)集

和join類似,但是這種交叉操作會產(chǎn)生笛卡爾積,在數(shù)據(jù)比較大的時(shí)候,是非常消耗內(nèi)存的操作

val cross = input1．cross(input2){
     (input1 , input2) => (input1．_1,input1．_2,input1．_3,input2．_2)
   }
cross．print()
14． union

聯(lián)合操作,創(chuàng)建包含來自該數(shù)據(jù)集和其他數(shù)據(jù)集的元素的新數(shù)據(jù)集,不會去重

val unionData: DataSet[String] = elements1．union(elements2)．union(elements3)
// 去除重復(fù)數(shù)據(jù)
val value = unionData．distinct(line => line)
15． rebalance

Flink也有數(shù)據(jù)傾斜的時(shí)候,比如當(dāng)前有數(shù)據(jù)量大概10億條數(shù)據(jù)需要處理,在處理過程中可能會發(fā)生如圖所示的狀況:

這個(gè)時(shí)候本來總體數(shù)據(jù)量只需要10分鐘解決的問題,出現(xiàn)了數(shù)據(jù)傾斜,機(jī)器1上的任務(wù)需要4個(gè)小時(shí)才能完成,那么其他3臺機(jī)器執(zhí)行完畢也要等待機(jī)器1執(zhí)行完畢后才算整體將任務(wù)完成;所以在實(shí)際的工作中,出現(xiàn)這種情況比較好的解決方案就是接下來要介紹的—rebalance(內(nèi)部使用round robin方法將數(shù)據(jù)均勻打散。這對于數(shù)據(jù)傾斜時(shí)是很好的選擇。)

// 使用rebalance操作,避免數(shù)據(jù)傾斜
val rebalance = filterData．rebalance()
16． partitionByHash

按照指定的key進(jìn)行hash分區(qū)

val data = new mutable．MutableList[(Int, Long, String)]
data．+=((1, 1L, "Hi"))
data．+=((2, 2L, "Hello"))
data．+=((3, 2L, "Hello world"))
val collection = env．fromCollection(data)
val unique = collection．partitionByHash(1)．mapPartition{
 line =>
   line．map(x => (x．_1 , x．_2 , x．_3))
}
unique．writeAsText("hashPartition", WriteMode．NO_OVERWRITE)
env．execute()
17． partitionByRange

根據(jù)指定的key對數(shù)據(jù)集進(jìn)行范圍分區(qū)

val data = new mutable．MutableList[(Int, Long, String)]
data．+=((1, 1L, "Hi"))
data．+=((2, 2L, "Hello"))
data．+=((3, 2L, "Hello world"))
data．+=((4, 3L, "Hello world, how are you?"))
val collection = env．fromCollection(data)
val unique = collection．partitionByRange(x => x．_1)．mapPartition(line => line．map{
 x=>
   (x．_1 , x．_2 , x．_3)
})
unique．writeAsText("rangePartition", WriteMode．OVERWRITE)
env．execute()
18． sortPartition

根據(jù)指定的字段值進(jìn)行分區(qū)的排序

val data = new mutable．MutableList[(Int, Long, String)]
   data．+=((1, 1L, "Hi"))
   data．+=((2, 2L, "Hello"))
   data．+=((3, 2L, "Hello world"))
   data．+=((4, 3L, "Hello world, how are you?"))
val ds = env．fromCollection(data)
   val result = ds
     ．map { x => x }．setParallelism(2)
     ．sortPartition(1, Order．DESCENDING)//第一個(gè)參數(shù)代表按照哪個(gè)字段進(jìn)行分區(qū)
     ．mapPartition(line => line)
     ．collect()
println(result)
三、Sink算子1． collect

將數(shù)據(jù)輸出到本地集合

result．collect()
2． writeAsText

將數(shù)據(jù)輸出到文件

Flink支持多種存儲設(shè)備上的文件,包括本地文件,hdfs文件等

Flink支持多種文件的存儲格式,包括text文件,CSV文件等

// 將數(shù)據(jù)寫入本地文件
result．writeAsText("/data/a", WriteMode．OVERWRITE)
// 將數(shù)據(jù)寫入HDFS
result．writeAsText("hdfs://node01:9000/data/a", WriteMode．OVERWRITE)
DataStream流處理算子

和DataSet一樣,DataStream也包括一系列的Transformation操作

一、Source算子

Flink可以使用 StreamExecutionEnvironment．a(chǎn)ddSource(source) 來為我們的程序添加數(shù)據(jù)來源。
Flink 已經(jīng)提供了若干實(shí)現(xiàn)好了的 source functions,當(dāng)然我們也可以通過實(shí)現(xiàn) SourceFunction 來自定義非并行的source或者實(shí)現(xiàn) ParallelSourceFunction 接口或者擴(kuò)展 RichParallelSourceFunction 來自定義并行的 source。

Flink在流處理上的source和在批處理上的source基本一致。大致有4大類:

基于本地集合的source(Collection-based-source)基于文件的source(File-based-source)- 讀取文本文件,即符合 TextInputFormat 規(guī)范的文件,并將其作為字符串返回基于網(wǎng)絡(luò)套接字的source(Socket-based-source)- 從 socket 讀取。元素可以用分隔符切分。自定義的source(Custom-source)

下面使用addSource將Kafka數(shù)據(jù)寫入Flink為例:

如果需要外部數(shù)據(jù)源對接,可使用addSource,如將Kafka數(shù)據(jù)寫入Flink,先引入依賴:

將Kafka數(shù)據(jù)寫入Flink:

val properties = new Properties()
properties．setProperty("bootstrap．servers", "localhost:9092")
properties．setProperty("group．id", "consumer-group")
properties．setProperty("key．deserializer", "org．a(chǎn)pache．kafka．common．serialization．StringDeserializer")
properties．setProperty("value．deserializer", "org．a(chǎn)pache．kafka．common．serialization．StringDeserializer")
properties．setProperty("auto．offset．reset", "latest")
val source = env．a(chǎn)ddSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))

基于網(wǎng)絡(luò)套接字的:

val source = env．socketTextStream("IP", PORT)
二、Transform轉(zhuǎn)換算子1． map

將DataSet中的每一個(gè)元素轉(zhuǎn)換為另外一個(gè)元素

dataStream．map { x => x * 2 }
2． FlatMap

采用一個(gè)數(shù)據(jù)元并生成零個(gè),一個(gè)或多個(gè)數(shù)據(jù)元。將句子分割為單詞的flatmap函數(shù)

dataStream．flatMap { str => str．split(" ") }
3． Filter

計(jì)算每個(gè)數(shù)據(jù)元的布爾函數(shù),并保存函數(shù)返回true的數(shù)據(jù)元。過濾掉零值的過濾器

dataStream．filter { _ != 0 }
4． KeyBy

邏輯上將流分區(qū)為不相交的分區(qū)。具有相同Keys的所有記錄都分配給同一分區(qū)。在內(nèi)部,keyBy()是使用散列分區(qū)實(shí)現(xiàn)的。指定鍵有不同的方法。

此轉(zhuǎn)換返回KeyedStream,其中包括使用被Keys化狀態(tài)所需的KeyedStream。

dataStream．keyBy(0)
5． Reduce

被Keys化數(shù)據(jù)流上的“滾動”Reduce。將當(dāng)前數(shù)據(jù)元與最后一個(gè)Reduce的值組合并發(fā)出新值

keyedStream．reduce { _ + _ }  
6． Fold

具有初始值的被Keys化數(shù)據(jù)流上的“滾動”折疊。將當(dāng)前數(shù)據(jù)元與最后折疊的值組合并發(fā)出新值

val result: DataStream[String] =  keyedStream．fold("start")((str, i) => { str + "-" + i })
// 解釋:當(dāng)上述代碼應(yīng)用于序列(1,2,3,4,5)時(shí),輸出結(jié)果“start-1”,“start-1-2”,“start-1-2-3”,．．．
7． Aggregations

在被Keys化數(shù)據(jù)流上滾動聚合。min和minBy之間的差異是min返回最小值,而minBy返回該字段中具有最小值的數(shù)據(jù)元(max和maxBy相同)。

keyedStream．sum(0);
keyedStream．min(0);
keyedStream．max(0);
keyedStream．minBy(0);
keyedStream．maxBy(0);
8． Window

可以在已經(jīng)分區(qū)的KeyedStream上定義Windows。Windows根據(jù)某些特征(例如,在最后5秒內(nèi)到達(dá)的數(shù)據(jù))對每個(gè)Keys中的數(shù)據(jù)進(jìn)行分組。這里不再對窗口進(jìn)行詳解,有關(guān)窗口的完整說明,請查看這篇文章:Flink 中極其重要的 Time 與 Window 詳細(xì)解析

dataStream．keyBy(0)．window(TumblingEventTimeWindows．of(Time．seconds(5)));
9． WindowAll

Windows可以在常規(guī)DataStream上定義。Windows根據(jù)某些特征(例如,在最后5秒內(nèi)到達(dá)的數(shù)據(jù))對所有流事件進(jìn)行分組。

注意:在許多情況下,這是非并行轉(zhuǎn)換。所有記錄將收集在windowAll 算子的一個(gè)任務(wù)中。

dataStream．windowAll(TumblingEventTimeWindows．of(Time．seconds(5)))
10． Window Apply

將一般函數(shù)應(yīng)用于整個(gè)窗口。

注意:如果您正在使用windowAll轉(zhuǎn)換,則需要使用AllWindowFunction。

下面是一個(gè)手動求和窗口數(shù)據(jù)元的函數(shù)

windowedStream．a(chǎn)pply { WindowFunction }
allWindowedStream．a(chǎn)pply { AllWindowFunction }
11． Window Reduce

將函數(shù)縮減函數(shù)應(yīng)用于窗口并返回縮小的值

windowedStream．reduce { _ + _ }
12． Window Fold

將函數(shù)折疊函數(shù)應(yīng)用于窗口并返回折疊值

val result: DataStream[String] = windowedStream．fold("start", (str, i) => { str + "-" + i })
// 上述代碼應(yīng)用于序列(1,2,3,4,5)時(shí),將序列折疊為字符串“start-1-2-3-4-5”
13． Union

兩個(gè)或多個(gè)數(shù)據(jù)流的聯(lián)合,創(chuàng)建包含來自所有流的所有數(shù)據(jù)元的新流。注意:如果將數(shù)據(jù)流與自身聯(lián)合,則會在結(jié)果流中獲取兩次數(shù)據(jù)元

dataStream．union(otherStream1, otherStream2, ．．．)
14． Window Join

在給定Keys和公共窗口上連接兩個(gè)數(shù)據(jù)流

dataStream．join(otherStream)
   ．where(

在給定的時(shí)間間隔內(nèi)使用公共Keys關(guān)聯(lián)兩個(gè)被Key化的數(shù)據(jù)流的兩個(gè)數(shù)據(jù)元e1和e2,以便e1．timestamp + lowerBound <= e2．timestamp <= e1．timestamp + upperBound

am．intervalJoin(otherKeyedStream)
   ．between(Time．milliseconds(-2), Time．milliseconds(2))
   ．upperBoundExclusive(true)
   ．lowerBoundExclusive(true)
   ．process(new IntervalJoinFunction() {．．．})
16． Window CoGroup

在給定Keys和公共窗口上對兩個(gè)數(shù)據(jù)流進(jìn)行Cogroup

dataStream．coGroup(otherStream)
   ．where(0)．equalTo(1)
   ．window(TumblingEventTimeWindows．of(Time．seconds(3)))
   ．a(chǎn)pply (new CoGroupFunction () {．．．})
17． Connect

“連接”兩個(gè)保存其類型的數(shù)據(jù)流。連接允許兩個(gè)流之間的共享狀態(tài)

DataStream

類似于連接數(shù)據(jù)流上的map和flatMap

connectedStreams．map(
   (_ : Int) => true,
   (_ : String) => false)connectedStreams．flatMap(
   (_ : Int) => true,
   (_ : String) => false)
19． Split

根據(jù)某些標(biāo)準(zhǔn)將流拆分為兩個(gè)或更多個(gè)流

val split = someDataStream．split(
 (num: Int) =>
   (num % 2) match {
     case 0 => List("even")
     case 1 => List("odd")
   })      
20． Select

從拆分流中選擇一個(gè)或多個(gè)流

SplitStream

支持將數(shù)據(jù)輸出到:

本地文件(參考批處理)本地集合(參考批處理)HDFS(參考批處理)

除此之外,還支持:

sink到kafkasink到mysqlsink到redis

下面以sink到kafka為例:

val sinkTopic = "test"
//樣例類
case class Student(id: Int, name: String, addr: String, sex: String)
val mapper: ObjectMapper = new ObjectMapper()
//將對象轉(zhuǎn)換成字符串
def toJsonString(T: Object): String = {
   mapper．registerModule(DefaultScalaModule)
   mapper．writeValueAsString(T)
}
def main(args: Array[String]): Unit = {
   //1．創(chuàng)建流執(zhí)行環(huán)境
   val env = StreamExecutionEnvironment．getExecutionEnvironment
   //2．準(zhǔn)備數(shù)據(jù)
   val dataStream: DataStream[Student] = env．fromElements(
     Student(8, "xiaoming", "beijing biejing", "female")
   )
   //將student轉(zhuǎn)換成字符串
   val studentStream: DataStream[String] = dataStream．map(student =>
     toJsonString(student) // 這里需要顯示SerializerFeature中的某一個(gè),否則會報(bào)同時(shí)匹配兩個(gè)方法的錯(cuò)誤
   )
   //studentStream．print()
   val prop = new Properties()
   prop．setProperty("bootstrap．servers", "node01:9092")
   val myProducer = new FlinkKafkaProducer011[String](sinkTopic, new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()), prop)
   studentStream．a(chǎn)ddSink(myProducer)
   studentStream．print()
   env．execute("Flink add sink")
}

五、流處理中的Time與Window

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

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

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

實(shí)時(shí):就是數(shù)據(jù)發(fā)送過來之后立馬就進(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)瀏覽新聞的用戶都是來自哪些地區(qū),這種就需要將五分鐘內(nèi)瀏覽新聞的用戶信息與 hive 中的地區(qū)維表進(jìn)行關(guān)聯(lián),然后在進(jìn)行相關(guān)計(jì)算。

本節(jié)所講的 Flink 內(nèi)容就是圍繞以上概念進(jìn)行詳細(xì)剖析的!

1． Time

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

Flink中的時(shí)間有三種類型,如下圖所示:

Event Time:是事件創(chuàng)建的時(shí)間。它通常由事件中的時(shí)間戳描述,例如采集的日志數(shù)據(jù)中,每一條日志都會記錄自己的生成時(shí)間,Flink通過時(shí)間戳分配器訪問事件時(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

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

2． Window

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

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

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

Window類型

本文剛開始提到,劃分窗口就兩種方式:

根據(jù)時(shí)間進(jìn)行截取(time-driven-window),比如每1分鐘統(tǒng)計(jì)一次或每10分鐘統(tǒng)計(jì)一次。根據(jù)數(shù)據(jù)進(jìn)行截取(data-driven-window),比如每5個(gè)數(shù)據(jù)統(tǒng)計(jì)一次或每50個(gè)數(shù)據(jù)統(tǒng)計(jì)一次。

窗口類型

對于TimeWindow(根據(jù)時(shí)間劃分窗口), 可以根據(jù)窗口實(shí)現(xiàn)原理的不同分成三類:滾動窗口(Tumbling Window)、滑動窗口(Sliding Window)和會話窗口(Session Window)。

滾動窗口(Tumbling Windows)

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

特點(diǎn):時(shí)間對齊,窗口長度固定,沒有重疊。

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

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

滾動窗口

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

滑動窗口(Sliding Windows)

滑動窗口是固定窗口的更廣義的一種形式,滑動窗口由固定的窗口長度和滑動間隔組成。

特點(diǎn):時(shí)間對齊,窗口長度固定,有重疊。

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

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

滑動窗口

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

會話窗口(Session Windows)

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

特點(diǎn):時(shí)間無對齊。

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

會話窗口3． Window API1) TimeWindow

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

我們以 紅綠燈路口通過的汽車數(shù)量 為例子:

紅綠燈路口會有汽車通過,一共會有多少汽車通過,無法計(jì)算。因?yàn)檐嚵髟丛床粩?計(jì)算沒有邊界。

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

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

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

1．開啟發(fā)送端口,端口號為9999
nc -lk 9999
2．發(fā)送內(nèi)容(key 代表不同的路口,value 代表每次通過的車輛)
一次發(fā)送一行,發(fā)送的時(shí)間間隔代表汽車經(jīng)過的時(shí)間間隔
9,3
9,2
9,7
4,9
2,6
1,5
2,3
5,7
5,4

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

object Window {
 def main(args: Array[String]): Unit = {
   //TODO time-window
   //1．創(chuàng)建運(yùn)行環(huán)境
   val env = StreamExecutionEnvironment．getExecutionEnvironment
   //2．定義數(shù)據(jù)流來源
   val text = env．socketTextStream("localhost", 9999)
   //3．轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
   case class CarWc(sensorId: Int, carCnt: Int)
   val ds1: DataStream[CarWc] = text．map {
     line => {
       val tokens = line．split(",")
       CarWc(tokens(0)．trim．toInt, tokens(1)．trim．toInt)
     }
   }
   //4．執(zhí)行統(tǒng)計(jì)操作,每個(gè)sensorId一個(gè)tumbling窗口,窗口的大小為5秒
   //也就是說,每5秒鐘統(tǒng)計(jì)一次,在這過去的5秒鐘內(nèi),各個(gè)路口通過紅綠燈汽車的數(shù)量。
   val ds2: DataStream[CarWc] = ds1
     ．keyBy("sensorId")
     ．timeWindow(Time．seconds(5))
     ．sum("carCnt")
   //5．顯示統(tǒng)計(jì)結(jié)果
   ds2．print()
   //6．觸發(fā)流計(jì)算
   env．execute(this．getClass．getName)
 }
}

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

sliding-time-window (有重疊數(shù)據(jù))//1．創(chuàng)建運(yùn)行環(huán)境
val env = StreamExecutionEnvironment．getExecutionEnvironment
//2．定義數(shù)據(jù)流來源
val text = env．socketTextStream("localhost", 9999)
//3．轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
case class CarWc(sensorId: Int, carCnt: Int)
val ds1: DataStream[CarWc] = text．map {
 line => {
   val tokens = line．split(",")
   CarWc(tokens(0)．trim．toInt, tokens(1)．trim．toInt)
 }
}
//4．執(zhí)行統(tǒng)計(jì)操作,每個(gè)sensorId一個(gè)sliding窗口,窗口時(shí)間10秒,滑動時(shí)間5秒
//也就是說,每5秒鐘統(tǒng)計(jì)一次,在這過去的10秒鐘內(nèi),各個(gè)路口通過紅綠燈汽車的數(shù)量。
val ds2: DataStream[CarWc] = ds1
 ．keyBy("sensorId")
 ．timeWindow(Time．seconds(10), Time．seconds(5))
 ．sum("carCnt")
//5．顯示統(tǒng)計(jì)結(jié)果
ds2．print()
//6．觸發(fā)流計(jì)算
env．execute(this．getClass．getName)
2) CountWindow

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

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

tumbling-count-window (無重疊數(shù)據(jù))//1．創(chuàng)建運(yùn)行環(huán)境
val env = StreamExecutionEnvironment．getExecutionEnvironment
//2．定義數(shù)據(jù)流來源
val text = env．socketTextStream("localhost", 9999)
//3．轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
case class CarWc(sensorId: Int, carCnt: Int)
val ds1: DataStream[CarWc] = text．map {
 (f) => {
   val tokens = f．split(",")
   CarWc(tokens(0)．trim．toInt, tokens(1)．trim．toInt)
 }
}
//4．執(zhí)行統(tǒng)計(jì)操作,每個(gè)sensorId一個(gè)tumbling窗口,窗口的大小為5
//按照key進(jìn)行收集,對應(yīng)的key出現(xiàn)的次數(shù)達(dá)到5次作為一個(gè)結(jié)果
val ds2: DataStream[CarWc] = ds1
 ．keyBy("sensorId")
 ．countWindow(5)
 ．sum("carCnt")
//5．顯示統(tǒng)計(jì)結(jié)果
ds2．print()
//6．觸發(fā)流計(jì)算
env．execute(this．getClass．getName)
sliding-count-window (有重疊數(shù)據(jù))

同樣也是窗口長度和滑動窗口的操作:窗口長度是5,滑動長度是3

//1．創(chuàng)建運(yùn)行環(huán)境
val env = StreamExecutionEnvironment．getExecutionEnvironment
//2．定義數(shù)據(jù)流來源
val text = env．socketTextStream("localhost", 9999)
//3．轉(zhuǎn)換數(shù)據(jù)格式,text->CarWc
case class CarWc(sensorId: Int, carCnt: Int)
val ds1: DataStream[CarWc] = text．map {
 (f) => {
   val tokens = f．split(",")
   CarWc(tokens(0)．trim．toInt, tokens(1)．trim．toInt)
 }
}
//4．執(zhí)行統(tǒng)計(jì)操作,每個(gè)sensorId一個(gè)sliding窗口,窗口大小3條數(shù)據(jù),窗口滑動為3條數(shù)據(jù)
//也就是說,每個(gè)路口分別統(tǒng)計(jì),收到關(guān)于它的3條消息時(shí)統(tǒng)計(jì)在最近5條消息中,各自路口通過的汽車數(shù)量
val ds2: DataStream[CarWc] = ds1
 ．keyBy("sensorId")
 ．countWindow(5, 3)
 ．sum("carCnt")
//5．顯示統(tǒng)計(jì)結(jié)果
ds2．print()
//6．觸發(fā)流計(jì)算
env．execute(this．getClass．getName)
Window 總結(jié)

flink支持兩種劃分窗口的方式(time和count)

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

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

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

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

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

如果size

通過組合可以得出四種基本窗口

time-tumbling-window 無重疊數(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無重疊數(shù)據(jù)的數(shù)量窗口,設(shè)置方式舉例:countWindow(5)

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

3) Window Reduce

WindowedStream → DataStream:給window賦一個(gè)reduce功能的函數(shù),并返回一個(gè)聚合的結(jié)果。

import org．a(chǎn)pache．flink．streaming．a(chǎn)pi．scala．StreamExecutionEnvironment
import org．a(chǎn)pache．flink．a(chǎn)pi．scala．_
import org．a(chǎn)pache．flink．streaming．a(chǎn)pi．windowing．time．Time
object StreamWindowReduce {
 def main(args: Array[String]): Unit = {
   // 獲取執(zhí)行環(huán)境
   val env = StreamExecutionEnvironment．getExecutionEnvironment
   // 創(chuàng)建SocketSource
   val stream = env．socketTextStream("node01", 9999)
   // 對stream進(jìn)行處理并按key聚合
   val streamKeyBy = stream．map(item => (item, 1))．keyBy(0)
   // 引入時(shí)間窗口
   val streamWindow = streamKeyBy．timeWindow(Time．seconds(5))
   // 執(zhí)行聚合操作
   val streamReduce = streamWindow．reduce(
     (item1, item2) => (item1．_1, item1．_2 + item2．_2)
   )
   // 將聚合數(shù)據(jù)寫入文件
   streamReduce．print()
   // 執(zhí)行程序
   env．execute("TumblingWindow")
 }
}
4) Window Apply

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

用法

實(shí)現(xiàn)一個(gè) WindowFunction 類指定該類的泛型為 [輸入數(shù)據(jù)類型, 輸出數(shù)據(jù)類型, keyBy中使用分組字段的類型, 窗口類型]

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

步驟:

獲取流處理運(yùn)行環(huán)境構(gòu)建socket流數(shù)據(jù)源,并指定IP地址和端口號對接收到的數(shù)據(jù)轉(zhuǎn)換成單詞元組使用 keyBy 進(jìn)行分流(分組)使用 timeWinodw 指定窗口的長度(每3秒計(jì)算一次)實(shí)現(xiàn)一個(gè)WindowFunction匿名內(nèi)部類apply方法中實(shí)現(xiàn)聚合計(jì)算使用Collector．collect收集數(shù)據(jù)

核心代碼如下:

   //1． 獲取流處理運(yùn)行環(huán)境
   val env = StreamExecutionEnvironment．getExecutionEnvironment
   //2． 構(gòu)建socket流數(shù)據(jù)源,并指定IP地址和端口號
   val textDataStream = env．socketTextStream("node01", 9999)．flatMap(_．split(" "))
   //3． 對接收到的數(shù)據(jù)轉(zhuǎn)換成單詞元組
   val wordDataStream = textDataStream．map(_->1)
   //4． 使用 keyBy 進(jìn)行分流(分組)
   val groupedDataStream: KeyedStream[(String, Int), String] = wordDataStream．keyBy(_．_1)
   //5． 使用 timeWinodw 指定窗口的長度(每3秒計(jì)算一次)
   val windowDataStream: WindowedStream[(String, Int), String, TimeWindow] = groupedDataStream．timeWindow(Time．seconds(3))
   //6． 實(shí)現(xiàn)一個(gè)WindowFunction匿名內(nèi)部類
   val reduceDatStream: DataStream[(String, Int)] = windowDataStream．a(chǎn)pply(new RichWindowFunction[(String, Int), (String, Int), String, TimeWindow] {
     //在apply方法中實(shí)現(xiàn)數(shù)據(jù)的聚合
     override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
       println("hello world")
       val tuple = input．reduce((t1, t2) => {
         (t1．_1, t1．_2 + t2．_2)
       })
       //將要返回的數(shù)據(jù)收集起來,發(fā)送回去
       out．collect(tuple)
     }
   })
   reduceDatStream．print()
   env．execute()
5) Window Fold

WindowedStream → DataStream:給窗口賦一個(gè)fold功能的函數(shù),并返回一個(gè)fold后的結(jié)果。

import org．a(chǎn)pache．flink．streaming．a(chǎn)pi．scala．StreamExecutionEnvironment
import org．a(chǎn)pache．flink．a(chǎn)pi．scala．_
import org．a(chǎn)pache．flink．streaming．a(chǎn)pi．windowing．time．Time
object StreamWindowFold {
 def main(args: Array[String]): Unit = {
   // 獲取執(zhí)行環(huán)境
   val env = StreamExecutionEnvironment．getExecutionEnvironment
   // 創(chuàng)建SocketSource
   val stream = env．socketTextStream("node01", 9999,'',3)
   // 對stream進(jìn)行處理并按key聚合
   val streamKeyBy = stream．map(item => (item, 1))．keyBy(0)
   // 引入滾動窗口
   val streamWindow = streamKeyBy．timeWindow(Time．seconds(5))
   // 執(zhí)行fold操作
   val streamFold = streamWindow．fold(100){
     (begin, item) =>
       begin + item．_2
   }
   // 將聚合數(shù)據(jù)寫入文件
   streamFold．print()
   // 執(zhí)行程序
   env．execute("TumblingWindow")
 }
}
6) Aggregation on Window

WindowedStream → DataStream:對一個(gè)window內(nèi)的所有元素做聚合操作。min和 minBy的區(qū)別是min返回的是最小值,而minBy返回的是包含最小值字段的元素(同樣的原理適用于 max 和 maxBy)。

import org．a(chǎn)pache．flink．streaming．a(chǎn)pi．scala．StreamExecutionEnvironment
import org．a(chǎn)pache．flink．streaming．a(chǎn)pi．windowing．time．Time
import org．a(chǎn)pache．flink．a(chǎn)pi．scala．_
object StreamWindowAggregation {
 def main(args: Array[String]): Unit = {
   // 獲取執(zhí)行環(huán)境
   val env = StreamExecutionEnvironment．getExecutionEnvironment
   // 創(chuàng)建SocketSource
   val stream = env．socketTextStream("node01", 9999)
   // 對stream進(jìn)行處理并按key聚合
   val streamKeyBy = stream．map(item => (item．split(" ")(0), item．split(" ")(1)))．keyBy(0)
   // 引入滾動窗口
   val streamWindow = streamKeyBy．timeWindow(Time．seconds(5))
   // 執(zhí)行聚合操作
   val streamMax = streamWindow．max(1)
   // 將聚合數(shù)據(jù)寫入文件
   streamMax．print()
   // 執(zhí)行程序
   env．execute("TumblingWindow")
 }
}
4． EventTime與Window1) EventTime的引入與現(xiàn)實(shí)世界中的時(shí)間是不一致的,在flink中被劃分為事件時(shí)間,提取時(shí)間,處理時(shí)間三種。如果以EventTime為基準(zhǔn)來定義時(shí)間窗口那將形成EventTimeWindow,要求消息本身就應(yīng)該攜帶EventTime如果以IngesingtTime為基準(zhǔn)來定義時(shí)間窗口那將形成IngestingTimeWindow,以source的systemTime為準(zhǔn)。如果以ProcessingTime基準(zhǔn)來定義時(shí)間窗口那將形成ProcessingTimeWindow,以operator的systemTime為準(zhǔn)。

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

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

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

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

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

Watermark 是用于處理亂序事件的,而正確的處理亂序事件,通常用 Watermark 機(jī)制結(jié)合 Window 來實(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í)長 t,每次系統(tǒng)會校驗(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)

有序數(shù)據(jù)的Watermark

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

無序數(shù)據(jù)的Watermark

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

上圖中,我們設(shè)置的允許最大延遲到達(dá)時(shí)間為2s,所以時(shí)間戳為7s的事件對應(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。

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

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

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

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

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

具體的用法如下:

allowedLateness(lateness: Time)

def allowedLateness(lateness: Time): WindowedStream[T, K, W] = {
 javaStream．a(chǎn)llowedLateness(lateness)
 this
}

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

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

隨著新數(shù)據(jù)的到來,waterMark的值會更新為最新數(shù)據(jù)事件時(shí)間-允許亂序時(shí)間值,但是如果這時(shí)候來了一條歷史數(shù)據(jù),waterMark值則不會更新。總的來說,waterMark是為了能接收到盡可能多的亂序數(shù)據(jù)。

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

注意:該方法只針對于基于event-time的窗口,如果是基于processing-time,并且指定了非零的time值則會拋出異常。

sideOutputLateData(outputTag: OutputTag[T])

def sideOutputLateData(outputTag: OutputTag[T]): WindowedStream[T, K, W] = {
 javaStream．sideOutputLateData(outputTag)
 this
}

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

DataStream．getSideOutput(tag: OutputTag[X])

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

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

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

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

那么問題來了,延遲時(shí)間怎么計(jì)算呢?

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