Flink Watermark和时间语义

Flink 中的时间语义

时间语义： EventTime：变乱创建时间；Ingestion Time：数据进入Flink的时间；Processing Time：实利用用算子的当地体系时间，与呆板无关。差异的时间语义有差异的应用场合，我们每每更关系变乱时间Event Time。数据天生的时间就会主动注入时间戳，Event Time可以从日记数据的时间戳timestamp)中提取。
设置 Event Time

我们可以直接在代码中，对实行环境调用setStreamTimeCharacteristic方法，设置流的时间特性。详细的时间，还须要从数据中提取时间戳timestamp。

1. val env = StreamExecutionEnvironment.getExecutionEnvironment
2. //从调用时刻开始给 env 创建的每一个 stream 追加时间特性
3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

复制代码

乱序数据的影响

当Flink以Event Time模式处理惩罚数据流时，它会根据数据里的时间戳来处理惩罚基于时间的算子。由于网络、分布式等缘故原由，会导致乱序数据的产生。如上图所示，抱负环境与实际环境会存在差异，乱序数据会让窗口盘算不精确。办理方案是让窗口等几分钟。
水位线 Watermark

怎么克制乱序数据带来盘算不精确？
遇到一个时间戳到达了窗口关闭时间，不应该立即触发窗口盘算，而是期待一段时间，等迟到的数据来了再关闭窗口。Watermark是一种权衡Event Time渴望的机制，可以设置延伸触发。Watermark是用于处理惩罚乱序变乱的，而精确的处理惩罚乱序变乱，通常用Watermark机制联合window来实现。数据流中的Watermark用于体现timestamp小于Watermark的数据，都已经到达了，因此，window的实行也是由Watermark触发的。Watermark用来让步伐本身延伸和效果精确性。
Watermark 的特点： Watermark是一条特别的数据记载，必须单调递增，以确保使命的变乱时间时钟在向前推进，而不是在退却。Watermark与数据的时间戳有关。

watermark 的通报、引入和设定

watermark的通报： 一个Task输入可以并行多个，如下有4个并行度，输出也大概存在多个并行，如下有3个。每个使命Task内部都有一个变乱时钟，且每个分区也维护了对应的WM，如下的Partition WM。当变乱流流进Partition时会判定新变乱流的WM是否大于当前的Partition WM，当大于时就更新Partition的时间戳WM为新流入的WM（取最大值），如下1->2象限Partition WM的厘革。同时，如下Task也维护了一个全局的WM体现变乱时钟，该值取分区中最小的WM作为输出的时间戳，如下第二象限的输出选择最小的WM=3向下通报。当第二个(横线)分区Partition WM流进来WM=7的变乱流时，就会出现第三象限的景象，但是最小的WM照旧=3，因此不更新Task全局的WM。当第三个分区Partition WM流进来WM=6的变乱流时，就会出现第四象限的景象，此时分区Partition WM的最小值=4，因此Task全局WM就=4。

watermark的引入： Event Time的利用肯定要指定命据源中的时间戳。对于排好序的数据，只须要指定时间戳就够了，不须要延伸触发。

1. import org.apache.flink.streaming.api.windowing.time.Time
2. //同时分配时间戳和水位线
3. dataStream.assignTimestampsAndWatermarks(
4. //无序数据       Time.milliseconds(1000)=延迟时间
5. new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.milliseconds(1000)) {
6.   //提取事件戳 = timestamp * 1000是因为出入的毫秒
7.   override def extractTimestamp(t: SensorReading): Long = {
8.     t.timestamp * 1000
9.   }
10. })

复制代码

【1】对于排好序的数据，不须要延伸触发，可以只指定变乱戳就行了

1. dataStream.assignTimestampsAndWatermarks(_.timestamp * 1000)

复制代码

【2】Flink袒露了TimestampAssigner接供词我们实现，使我们可以自界说怎样从变乱数据中抽取时间戳和天生 watermark。MyAssigner可以有两种范例，都继承自TimestampAssigner。

1. dataStream.assignTimestampsAndWatermarks(new MyAssigner())

复制代码

TimestampAssigner：界说了抽取时间戳，以及天生watermark的方法，有两种范例：
【1】AssignerWithPeriodicWatermarks： 体系会周期性的将Watermark插入到流中。默认周期是200毫秒（假如是processingTime则Watermark = 0 ），可以利用ExecutionConfig.setAutoWatermarkInterval()方法举行设置。升序和前面乱序的处理惩罚BoundedOutOfOrderness，都是基于周期性watermark的。举例：如下产生watermark的逻辑：每隔5秒，Flink调用AssignerWithPeriodicWatermarks的getCurrentWatermark()方法。假如方法返回一个时间戳大于之前水位的时间戳，新的water会被插入到流中。这个查抄包管了水位线是单调递增的。假如方法返回的时间戳小于之前水位的时间戳，则不会产生新的watermark。

1. //方案一:
2. //EventTime是以数据自带的时间戳字段为准，应用程序需要指定如何从record中抽取时间戳字段
3. env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
4. //每隔 5秒产生一个 watermark
5. env.getConfig.setAutoWatermarkInterval(5000);
7. ​
8. //方案二
9. //自定义一个周期性的时间戳
10. class PeriodicAssigner extends AssignerWithPeriodicWatermarks[SensorReading]{
12.   val bound: Long = 60 * 1000 //延时为 1 分钟
13.   var maxTs: Long = Long.MinValue //观察到的最大时间戳
15.   //生成水位线
16.   override def getCurrentWatermark: Watermark = {
17.     new Watermark(maxTs - bound)
18.   }
20.   //抽取时间戳的方法
21.   override def extractTimestamp(t: SensorReading, l: Long): Long = {
22.     maxTs = maxTs.max(t.timestamp)
23.     t.timestamp
24.   }
25. }

复制代码

【2】AssignerWithPunctuatedWatermarks： 没偶然间周期规律，可打断的天生watermark。

1. class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[SensorReading]{
2.   val bound: Long = 60 * 1000
4.   //获取水位线，根据数据触发
5.   override def checkAndGetNextWatermark(t: SensorReading, l: Long): Watermark = {
6.     if(t.id == "sensor_1"){
7.       new Watermark(l - bound)
8.     }else{
9.       null
10.     }
11.   }
13.   //抽取时间戳的方法
14.   override def extractTimestamp(t: SensorReading, l: Long): Long = {
15.     t.timestamp
16.   }
17. }

复制代码

watermark 的设定：
【1】在Flink中，watermark由应用步伐开辟职员天生，这通常须要对相应的范畴有肯定的相识。
【2】假如watermark设置的延伸太久，收到效果的速率大概就会很慢，办理办法是在水位线到达之前输出一个近似效果。
【3】而假如watermark到达得太早，则大概收到错误效果，不外Flink处理惩罚迟到数据的机制可以办理这个题目。