一、Time

在Flink的流式處理中，會涉及到時間的不同概念

Event Time：是事件創建的時間。它通常由事件中的時間戳描述，例如采集的日志數據中，每一條日志都會記錄自己的生成時間，Flink通過時間戳分配器訪問事件時間戳

Ingestion Time：是數據進入Flink的時間

Processing Time：是每一個執行基于時間操作的算子的本地系統時間，與機器相關，默認的時間屬性就是Processing Time。

例如一條日志進入Flink的時間為2017-11-12 10:00:00.123 到達window的系統時間為 2017-11-12 10:00:01.234，日志內容如下：

2017-11-02 18:37:15.624 INFO Fair over to rm2

對于業務來說，要統計1min內的故障日志個數，哪個時間是最有意義的？----- eventTime，因為我們要根據日志的生成時間進行統計。

??

如果要想聚合，不可能對無解數據流進行聚合。

?

二、Window

1、streaming流式計算是一種被設計用于處理處理無限數據集的數據處理引擎，而無限數據集是指一種不斷增長的本質上無限的數據集，而window是一種切割無限數據為有限塊進行處理的手段。

Window是無限數據流處理的核心，Window將一個無限的stream拆分成有限大小的"buckets"桶，我們可以在這些桶上做計算操作。

共有兩類，五種時間窗口。

2、Window類型（兩類）

2.1、CountWindow：按照指定的數據條數生成一個window，與時間無關

2.2、TimeWindow：按照時間生成window。（按照Processing Time來劃分Window）

對于TimeWindow和CountWindow，可以根據窗口實現原理的不同分成三類：滾動窗口（Tumbling Window）、滑動窗口（Sliding Window）和會話窗口（Session Window）。

（1）滾動窗口（Tumbling Windows）

將數據依據固定的窗口長度對數據進行切分。

特點：時間對齊，窗口長度固定，沒有重疊。

滾動窗口分配器將每個元素分配到一個指定窗口大小的窗口中，滾動窗口有一個固定的大小，并且不會出現重疊。

（2）滑動窗口（Sliding Windows）

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

特點：時間對齊，窗口長度固定，有重疊。

滑動窗口分配器將元素分配到固定長度的窗口中，與滾動窗口類似，窗口的大小由窗口大小參數來配置，另一個窗口滑動參數控制滑動窗口開始的頻率。

因此，滑動窗口如果滑動參數小于窗口大小的話，窗口是可以重疊的，在這種情況下元素會被分配到多個窗口中。

使用場景：對最近一個時間段內的統計（求某接口最近5min的失敗率來決定是否要報警。）

（3）會話窗口（Session Windows）

由一系列事件組合一個指定時間長度的timeout間隙組成。類似于web應用的session，也就是一段時間沒有接收到新數據就會生成新的窗口。

特點：時間無對齊。

session 窗口分配器通過session活動來對元素進行分組，session窗口跟滾動窗口和滑動窗口相比，不會有重疊和固定的開始時間和結束時間的情況，相反，當它在一個固定的

時間周期內不再收到元素，即非活動間隔產生，那這個窗口就會關閉。一個Session窗口通過一個session間隔來配置，這個session間隔定義了非活躍周期的長度，當這個非活躍

周期產生，那么當前的session將關閉并且后續的元素將被分配到新的session窗口中去。

?

三、Window API

3.1、CountWindow

CountWindow根據窗口中相同key元素的數量來觸發執行，執行時只計算元素數量達到窗口大小的key對應的結果。

注意：CountWindow的window_size 指的是相同key的元素的個數，不是輸入的所有元素的總數。

import org.apache.flink.api.java.tuple.Tuple import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}/*** CountWindow 中的滾動窗口（Tumbling Windows）* 將數據依據固定的窗口長度對數據進行切分。*/ object TimeAndWindow {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentval stream: DataStream[String] = env.socketTextStream("localhost",11111)val streamKeyBy: KeyedStream[(String, Long), Tuple] = stream.map(item => (item,1L)).keyBy(0)//注意：CountWindow的window_size 指的是相同key的元素的個數，不是輸入的所有元素的總數。val streamWindow: DataStream[(String, Long)] = streamKeyBy.countWindow(5).reduce((item1, item2)=>(item1._1,item1._2+item2._2))streamWindow.print()env.execute("TimeAndWindow")} }

3.2

import org.apache.flink.api.java.tuple.Tuple import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment}/*** CountWindow 中的滑動窗口（Sliding Windows）* 將數據依據固定的窗口長度對數據進行切分。*/ object TimeAndWindow {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentval stream: DataStream[String] = env.socketTextStream("localhost",11111)val streamKeyBy: KeyedStream[(String, Long), Tuple] = stream.map(item => (item,1L)).keyBy(0)//注意：CountWindow的window_size 指的是相同key的元素的個數，不是輸入的所有元素的總數。//滿足步長，就執行一次，按第一個參數的長度val streamWindow: DataStream[(String, Long)] = streamKeyBy.countWindow(5,2).reduce((item1, item2)=>(item1._1,item1._2+item2._2))streamWindow.print()env.execute("TimeAndWindow")} }

四、EventTime與Window

1、EventTime的引入

在Flink的流式處理中，絕大部分的業務都會使用eventTime，一般只在eventTime無法使用時，才會被迫使用ProcessingTime或者IngestionTime。

如果要使用EventTime，那么需要引入EventTime的時間戳，引入方式如下所示：

2、Watermark

　　概念：我們知道，流處理從事件產生，到流經source，再到operator，中間是有一個過程和時間的，雖然大部分情況下，流到operator的數據都是按照事件產生的

事件戳順序來的，但是也不排除由于網絡、背壓等原因，導致亂序的產生，所謂亂序，就是指Flink接收到的事件的先后順序不是嚴格按照事件的EventTime順序排列的。

　　Watermark是一種衡量Event Time進展的機制，它是數據本身的一個隱藏屬性，數據本身攜帶著對應的Watermark。

　　Watermark是用于處理亂序事件的，而正確的處理亂序事件，通常用Watermark機制結合window來實現。

　　數據流中的Watermark用于表示eventTime小于Watermark的數量，都已經到達了，因此，window的執行也是由Watermark觸發的。

　　Watermark可以理解成一個延遲觸發機制。我們可以設置Watermark的延時時長t，每次系統會校驗已經到達的數據中最大的maxEventTime，然后認定eventTime 小于

maxEventTime-t 的所有數據都已經到達。如果有窗口的停止時間等于maxEventTime-t，那么這個窗口被觸發執行。

滾動窗口/滑動窗口/會話窗口

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**
* TimeWindow
*/
object EventTimeAndWindow {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
//開啟watermark
//從調用時刻開始給env創建的每一個stream追加時間特征。
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val stream: KeyedStream[(String, Long), Tuple] = env.socketTextStream("192.168.218.130", 1111).assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor[String](Time.milliseconds(3000)) {
override def extractTimestamp(element: String): Long = {
// event word eventTime是日志生成時間，我們從日志中解析EventTime
val eventTime = element.split(" ")(0).toLong
println(eventTime)
eventTime
}
}
).map(item => (item.split(" ")(1),1L)).keyBy(0)
//加上滾動窗口，窗口大小是5s，調用window的api
// val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] = stream.window(TumblingEventTimeWindows.of(Time.seconds(5)))
//滑動窗口
// val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] = stream.window(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))
//會話窗口
val streamWindow: WindowedStream[(String, Long), Tuple, TimeWindow] = stream.window(EventTimeSessionWindows.withGap(Time.seconds(5)))
val streamReduce = streamWindow.reduce((item1,item2)=>(item1._1,item1._2+item2._2))
streamReduce.print()

env.execute("EventTimeAndWindow")
}
}

?

轉載于:https://www.cnblogs.com/ssqq5200936/p/11014296.html

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的Flink中的Time与Window的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。