Flink 四大基石之时间和水位线原理介绍！

- 日理万妓 2022-10-12 01:37 160阅读 0赞

原文链接：[Flink四大基石之时间和水位线原理介绍！][Flink]

在Flink中，涉及到时间和水位线这一概念，时间是**Flink中的四大基石（Checkpoint、State、Time、Window）**之一，是实现流批统一的一个重要特性。本文讲解内容包含以下六部分：

*  Flink 时间分类    
 *  WaterMark引入
 *  水印运行原理
 *  水印API调用
 *  侧道输出保障超过WaterMark数据不丢失 
 *  WaterMark+EventTimeWindow+Allowed Lateness案例

![20210702184137369.png][]

在Flink的流式处理中，会涉及到时间的不同概念，如下图所示：

![图片][4ebe49ee03845d13fae52fe8450d97eb.png]

**- EventTime\[事件时间\]**

事件发生的时间，例如：点击网站上的某个链接的时间，每一条日志都会记录自己的生成时间

如果以EventTime为基准来定义时间窗口那将形成EventTimeWindow,要求消息本身就应该携带EventTime

**- IngestionTime\[摄入时间\]**

数据进入Flink的时间，如某个Flink节点的sourceoperator接收到数据的时间，例如：某个source消费到kafka中的数据

如果以IngesingtTime为基准来定义时间窗口那将形成IngestingTimeWindow,以source的systemTime为准

**- ProcessingTime\[处理时间\]**

某个Flink节点执行某个operation的时间，例如：timeWindow处理数据时的系统时间，默认的时间属性就是Processing Time

如果以ProcessingTime基准来定义时间窗口那将形成ProcessingTimeWindow，以operator的systemTime为准

在Flink的流式处理中，绝大部分的业务都会使用**EventTime**，一般只在EventTime无法使用时，才会被迫使用ProcessingTime或者IngestionTime。

如果要使用EventTime，那么需要引入EventTime的时间属性，引入方式如下所示：

三种时间设置代码如下：

final StreamExecutionEnvironment env  
            = StreamExecutionEnvironment.getExecutionEnvironrnent();
    // 使用处理时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) ; 
    // 使用摄入时间
    
    env.setStrearnTimeCharacteristic(TimeCharacteristic.IngestionTime);
    
    // 使用事件时间
    env.setStrearnTimeCharacteristic(TimeCharacteri stic Eve~tTime);

![20210702184327229.png][]

通过两个案例来了解什么是水印，主要解决什么问题？

**案例1： **假你正在去往地下停车场的路上，并且打算用手机点一份外卖。

选好了外卖后，你就用在线支付功能付款了，这个时候是11点50分。恰好这时，你走进了地下停车库，而这里并没有手机信号。因此外卖的在线支付并没有立刻成功，而支付系统一直在Retry重试“支付”这个操作。

当你找到自己的车并且开出地下停车场的时候，已经是12点05分了。这个时候手机重新有了信号，手机上的支付数据成功发到了外卖在线支付系统，支付完成。

在上面这个场景中你可以看到，**支付数据的事件时间是11点50分**，而**支付数据的处理时间是12点05分**

**案例2：** 如上图所示，某App 会记录用户的所有点击行为，并回传日志（在网络不好的情况下，先保存在本地，延后回传）。

A 用户在11:02 对 App 进行操作，B用户在11:03 操作了 App，

但是A 用户的网络不太稳定，回传日志延迟了，导致我们在服务端先接受到B 用户11:03 的消息，然后再接受到A 用户11:02 的消息，**消息乱序**了。

通过上面的例子,我们知道,在进行数据处理的时候应该按照事件时间进行处理,也就是窗口应该要考虑到事件时间，但是窗口不能无限的一直等到延迟数据的到来,需要有一个触发窗口计算的机制，就是watermaker水位线/水印机制。

**所以：****水印是用来解决数据延迟、数据乱序等问题，总结如下图所示：**

![图片][5210cf8a0ae32225ff94847519158441.png]

水印就是一个时间戳（timestamp），Flink可以给数据流添加水印

- 水印并不会影响原有Eventtime事件时间

- 当数据流添加水印后，会按照水印时间来触发窗口计算,也就是说watermark水印是用来触发窗口计算的

- 设置水印时间，会比事件时间小几秒钟,表示最大允许数据延迟达到多久

- 水印时间 = 事件时间 - 允许延迟时间 (例如：10:09:57 =  10:10:00 - 3s )

![20210702184407906.png][]

如下图所示：

![图片][287bdee038aa0e2be7371d12fb72c29a.png]

窗口是10分钟触发一次，现在在12:00 - 12:10 有一个窗口，本来有一条数据是在12:00 - 12:10这个窗口被计算，但因为延迟，12:12到达，这时12:00 - 12:10 这个窗口就会被关闭，只能将数据下发到下一个窗口进行计算，这样就产生了数据延迟，造成计算不准确。

现在添加一个水位线：数据时间戳为2分钟。这时用数据产生的事件时间 12:12 -允许延迟的水印 2分钟 = 12:10 >= 窗口结束时间 。窗口触发计算，该数据就会被计算到这个窗口里

![202107021844296.png][]

在DataStream  API 中使用 TimestampAssigner 接口定义时间戳的提取行为，包含两个子接口 **AssignerWithPeriodicWatermarks **接口和** AssignerWithPunctuatedWaterMarks**接口

![图片][90d50cf035d21c57a2de62c86eedaeee.png]

![图片][12c948b0f942d9541474675e5e9006bb.png]

![20210702184458284.png][]

使用 WaterMark+ EventTimeWindow 机制可以在一定程度上解决数据乱序的问题，但是，WaterMark 水位线也不是万能的，在某些情况下，数据延迟会非常严重，即使通过Watermark + EventTimeWindow也无法等到数据全部进入窗口再进行处理，因为窗口触发计算后，**对于延迟到达的本属于该窗口的数据，Flink默认会将这些延迟严重的数据进行丢弃**

那么如果想要让一定时间范围的延迟数据不会被丢弃，可以使用Allowed Lateness(允许迟到机制/侧道输出机制)设定一个允许延迟的时间和侧道输出对象来解决

即使用**WaterMark + EventTimeWindow + Allowed Lateness方案**（包含侧道输出），可以做到数据不丢失。

**API调用**

l  allowedLateness(lateness:Time)---设置允许延迟的时间

该方法传入一个Time值，设置允许数据迟到的时间，这个时间和watermark中的时间概念不同。再来回顾一下，

**watermark=数据的事件时间-允许乱序时间值**

随着新数据的到来，watermark的值会更新为最新数据事件时间-允许乱序时间值，但是如果这时候来了一条历史数据，watermark值则不会更新。

**总的来说，watermark永远不会倒退它是为了能接收到尽可能多的乱序数据。**

那这里的Time值呢？主要是为了等待迟到的数据，如果属于该窗口的数据到来，仍会进行计算，后面会对计算方式仔细说明

注意：该方法只针对于基于event-time的窗口

l  sideOutputLateData(outputTag:OutputTag\[T\])--保存延迟数据

该方法是将迟来的数据保存至给定的outputTag参数，而OutputTag则是用来标记延迟数据的一个对象。

l  DataStream.getSideOutput(tag:OutputTag\[X\])--获取延迟数据

通过window等操作返回的DataStream调用该方法，传入标记延迟数据的对象来获取延迟的数据

![20210702184518845.png][]

package cn.itcast.watermark
    
    import org.apache.flink.api.scala._
    import org.apache.flink.streaming.api.TimeCharacteristic
    import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
    import org.apache.flink.streaming.api.scala.function.WindowFunction
    import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment, WindowedStream}
    import org.apache.flink.streaming.api.watermark.Watermark
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    /*
    演示水印如何解决数据乱序问题
     */
    object SlideOutputDemo {
      //样例类 CarWc(信号等id,数量)
      case class CarWc(id: String, num: Int, ts: Long)
      def main(args: Array[String]): Unit = {
        // 1 获取运行环境
        val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
        //方便观察数据设置并行度为1
        env.setParallelism(1)
        //设置基于事件时间进行计算
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        //2 接收数据
        val socketDs = env.socketTextStream("hlink161", 9999)
        // 3 transformation 接收到数据之后按照逗号切分，拿到红绿灯id,数量（通过汽车数量）
        val carWcDs: DataStream[CarWc] = socketDs.map(
          line => {
            val arr: Array[String] = line.split(",")
            // println("===="+line)
            //3.1 封装样例类
            CarWc(arr(0), arr(1).toInt, arr(2).toLong)
          }
        )
        // 3.1 设置水印机制,我们自己动手实现
        val waterMarkDs: DataStream[CarWc] = carWcDs.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[CarWc]{
          /** The current maximum timestamp seen so far. */
          private var currentMaxTimestamp = 0L
          //定义最大允许的延迟时间
          private var maxOutOfOrderness=0
    
          /** The timestamp of the last emitted watermark. */ //巨大的负数
          private var lastEmittedWatermark:Long= Long.MinValue
          //3.1.1计算并生成水印
          override def getCurrentWatermark: Watermark = {
            // this guarantees that the watermark never goes backwards.
            //计算出的watermark时间
            val potentialWM = currentMaxTimestamp - maxOutOfOrderness
            //保证水印时间不会回退！！
            if (potentialWM >= lastEmittedWatermark)
              lastEmittedWatermark = potentialWM
            return new Watermark(lastEmittedWatermark)
          }
          //3.1.2 获取到eventtime字段
          override def extractTimestamp(element: CarWc, previousElementTimestamp: Long): Long = {
            //获取日志的eventtime
            val timestamp = element.ts
            //判断比较 eventtime是不是大于最大的eventtime时间
            if (timestamp > currentMaxTimestamp)
              currentMaxTimestamp = timestamp
            //返回日志的eventtime时间
            timestamp
          }
        } )
    
        // pre-4 定义一个侧输出流
        val outputTag: OutputTag[CarWc] = new OutputTag[CarWc]("delayCarWc")
        // 4 设置窗口
    
        val windowStream: WindowedStream[CarWc, String, TimeWindow] = waterMarkDs.keyBy(_.id).timeWindow(Time.seconds(5))
          .allowedLateness(Time.seconds(5))
          .sideOutputLateData(outputTag)
    
    
        //5 使用apply方法执行聚合计算
        val windowRes: DataStream[CarWc] = windowStream.apply(new WindowFunction[CarWc, CarWc, String, TimeWindow] {
          override def apply(key: String, window: TimeWindow, input: Iterable[CarWc], out: Collector[CarWc]): Unit = {
            val iter: Iterator[CarWc] = input.iterator
            println("窗口开始时间 》》"+window.getStart+"== 窗口结束时间》》"+window.getEnd+"数据 【"+input.iterator.mkString(";")+"】")
            val wc: CarWc = iter.reduce((t1, t2) =>
              CarWc(t1.id, t1.num + t2.num, t1.ts)
            )
            out.collect(wc)
          }
        })
        windowRes.print("窗口计算结果>>")
        //获取到侧输出流结果
        val outputDs: DataStream[CarWc] = windowRes.getSideOutput(outputTag)
        outputDs.print("侧输出流数据")
        env.execute()
      }
    }

**以上就是时间及水位线原理讲解的所有内容！觉得好的，点赞，在看，分享三连击，谢谢！！！**

![20210702184952742.png][]

找各类大数据技术文章和面经，就来

<**3分钟秒懂大数据**>

随时更新互联网大数据组件内容

专为学习者提供技术博文

快和身边的小伙伴一起关注我们吧！

![图片][f3241ead7676baf5a1b8841321ac7407.png]

作者简介：逆流而上Mr李，秋招7offer,CSDN博客：https://blog.csdn.net/weixin\_38201936

[Flink]: https://mp.weixin.qq.com/s?__biz=MzU4NjY4NDU0MQ==&mid=2247489179&idx=1&sn=2cb22b2d50652f7e0bc07bf6cbf61489&chksm=fdf6d789ca815e9f76a48a08077a5ee712d822ae73ace496b6b65ae7eae2cab5ce40477cb889&token=1911354629&lang=zh_CN#rd
[20210702184137369.png]: /images/20221005/a989f1e70b2546bfacfec5336d35af71.png
[4ebe49ee03845d13fae52fe8450d97eb.png]: /images/20221005/8de21ae1c825411dbd000e501cdcee5e.png
[20210702184327229.png]: /images/20221005/b67e9257aacb4c24ba97a477c04edcea.png
[5210cf8a0ae32225ff94847519158441.png]: /images/20221005/2b801e05eea34a8589f81eee09539eea.png
[20210702184407906.png]: /images/20221005/4246d44f5595467fa41eb3150c004670.png
[287bdee038aa0e2be7371d12fb72c29a.png]: /images/20221005/90b7c5794d5d475fbefc461055d6c73e.png
[202107021844296.png]: /images/20221005/bcfb1441942a44fab6deb3e89e532a21.png
[90d50cf035d21c57a2de62c86eedaeee.png]: /images/20221005/61e16189ab1d473cbd28060cb67804a1.png
[12c948b0f942d9541474675e5e9006bb.png]: /images/20221005/0c1eee6b0e7a4343ad62ea72c0c649b1.png
[20210702184458284.png]: /images/20221005/830bfd7b63a44d2292570750689d4c76.png
[20210702184518845.png]: /images/20221005/6eef9fbda6b440acbfb933731ba84a33.png
[20210702184952742.png]: /images/20221005/743def23343a47f389829d43609d03fa.png
[f3241ead7676baf5a1b8841321ac7407.png]: /images/20221005/09f6071252c348f98464d48f1c16e670.png