【Flink】Flink的事件时间及watermark水印讲解-附代码实例

谁践踏了优雅 2022-04-25 07:24 449阅读 0赞

代码链接：[https://github.com/18113996630/flink-practice/blob/master/src/main/scala/com/hrong/flink/watermark/WaterMarkFunc01.scala][https_github.com_18113996630_flink-practice_blob_master_src_main_scala_com_hrong_flink_watermark_WaterMarkFunc01.scala]

视频讲解：[flink水印视频讲解][flink]

Flink支持根据事件时间处理，数据流中的每条数据都需要具有各自的时间戳，代表着数据的产生时间【事件时间】。

在分布式系统中，数据流的采集通常都是有延迟的，可能是网络原因啊，程序原因啊什么的。所以当数据到达Flink程序中的时候，问题就来了，这些数据都要进行处理吗？有可能其中一部分数据已经延迟了好几个小时了，这对于实时性较强的业务场景是不能容忍的！

这时候水印就应运而生了，水印的目的就是为了解决乱序的数据问题，可以在时间窗口内根据事件时间来进行业务处理，对于乱序的有延迟的数据可以在一定时间范围内进行等待，那这个时间范围是怎么计算的呢？

--------------------

### 我们先来捋捋思路 ###

数据在源源不断的进入flink，我们设置好window的大小为5s，flink会以5s来将每分钟划分为连续的多个窗口

则flink划分的时间窗口为(**左闭右开**):

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td style="text-align:center;vertical-align:middle;"><strong>窗口起始时间</strong></td> 
   <td style="text-align:center;vertical-align:middle;"><strong>窗口结束时间</strong></td> 
  </tr> 
  <tr> 
   <td style="text-align:center;vertical-align:middle;">0</td> 
   <td style="text-align:center;vertical-align:middle;">5</td> 
  </tr> 
  <tr> 
   <td style="text-align:center;vertical-align:middle;">5</td> 
   <td style="text-align:center;vertical-align:middle;">10</td> 
  </tr> 
  <tr> 
   <td style="text-align:center;vertical-align:middle;">10</td> 
   <td style="text-align:center;vertical-align:middle;">15</td> 
  </tr> 
  <tr> 
   <td style="text-align:center;vertical-align:middle;">...</td> 
   <td style="text-align:center;vertical-align:middle;">...</td> 
  </tr> 
  <tr> 
   <td style="text-align:center;vertical-align:middle;">55</td> 
   <td style="text-align:center;vertical-align:middle;">60</td> 
  </tr> 
 </tbody> 
</table>

进入flink的第一条数据会落在一个时间窗口内，假设数据的事件时间为13s(小时和分不重要，因为窗口大小的度量单位是秒)，则落入的窗口是【10-15】。对于存在延迟的数据，我们能容忍的时间是3s，超过3s我就不等你了，继续进行窗口操作。

这里就要提到一个知识点：**Window的触发条件是什么，什么时候开始进行window操作？**

*  **该窗口中存在数据**
 *  **事件时间水印值到达窗口的结束时间**

好，知道了window触发条件后我们继续分析

第一个条件肯定满足的，只要有数据就行了

第二个条件，窗口的结束时间是15s，但是我们加了水印，允许数据延迟3秒，换句话说就是本来在15秒这个窗口就应该开始统计数据了，但是为了等一些延迟的数据，我要在18s才开始进行统计

【10-15】窗口触发的条件就是：***存在一条数据的事件时间大于等于18s***

### 下面我们用实例来验证： ###

大概讲解一下代码的流程：

1、监听某主机的9000端口，读取socket数据(格式为  name:timestamp)

2、给当前进入flink程序的数据加上waterMark，值为eventTime-3s

3、根据name值进行分组，根据窗口大小为5s划分窗口，依次统计窗口中各name值的数据

4、启动Job

下面是具体的代码：

import org.apache.commons.lang3.time.FastDateFormat
    import org.apache.flink.api.java.tuple.Tuple
    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, StreamExecutionEnvironment}
    import org.apache.flink.streaming.api.watermark.Watermark
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
    import org.apache.flink.streaming.api.windowing.time.Time
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow
    import org.apache.flink.util.Collector
    
    import scala.collection.mutable.ArrayBuffer
    
    /**
      * 水印测试
      */
    object WaterMarkFunc01 {
      // 线程安全的时间格式化对象
      val sdf: FastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss:SSS")
    
      def main(args: Array[String]): Unit = {
        val hostName = "s102"
        val port = 9000
        val delimiter = '\n'
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        // 将EventTime设置为流数据时间类型
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
        env.setParallelism(1)
        val streams: DataStream[String] = env.socketTextStream(hostName, port, delimiter)
        import org.apache.flink.api.scala._
        val data = streams.map(data => {
          // 输入数据格式：name:时间戳
          // flink:1559223685000
          try {
            val items = data.split(":")
            (items(0), items(1).toLong)
          } catch {
            case _: Exception => println("输入数据不符合格式：" + data)
              ("0", 0L)
          }
        }).filter(data => !data._1.equals("0") && data._2 != 0L)
    
        //为数据流中的元素分配时间戳，并定期创建水印以监控事件时间进度
        val waterStream: DataStream[(String, Long)] = data.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[(String, Long)] {
          // 事件时间
          var currentMaxTimestamp = 0L
          val maxOutOfOrderness = 3000L
          var lastEmittedWatermark: Long = Long.MinValue
    
          // Returns the current watermark
          override def getCurrentWatermark: Watermark = {
            // 允许延迟三秒
            val potentialWM = currentMaxTimestamp - maxOutOfOrderness
            // 保证水印能依次递增
            if (potentialWM >= lastEmittedWatermark) {
              lastEmittedWatermark = potentialWM
            }
            new Watermark(lastEmittedWatermark)
          }
    
          // Assigns a timestamp to an element, in milliseconds since the Epoch
          override def extractTimestamp(element: (String, Long), previousElementTimestamp: Long): Long = {
            // 将元素的时间字段值作为该数据的timestamp
            val time = element._2
            if (time > currentMaxTimestamp) {
              currentMaxTimestamp = time
            }
            val outData = String.format("key: %s    EventTime: %s    waterMark:  %s",
              element._1,
              sdf.format(time),
              sdf.format(getCurrentWatermark.getTimestamp))
            println(outData)
            time
          }
        })
        val result: DataStream[String] = waterStream.keyBy(0)// 根据name值进行分组
          .window(TumblingEventTimeWindows.of(Time.seconds(5L)))// 5s跨度的基于事件时间的翻滚窗口
          .apply(new WindowFunction[(String, Long), String, Tuple, TimeWindow] {
            override def apply(key: Tuple, window: TimeWindow, input: Iterable[(String, Long)], out: Collector[String]): Unit = {
              val timeArr = ArrayBuffer[String]()
              val iterator = input.iterator
              while (iterator.hasNext) {
                val tup2 = iterator.next()
                timeArr.append(sdf.format(tup2._2))
              }
              val outData = String.format("key: %s    data: %s    startTime:  %s    endTime:  %s",
                key.toString,
                timeArr.mkString("-"),
                sdf.format(window.getStart),
                sdf.format(window.getEnd))
              out.collect(outData)
            }
          })
        result.print("window计算结果:")
        env.execute(this.getClass.getName)
      }
    }

还记得我们开始说的吗

flink会根据window的间隔时间进行时间窗口范围的划分(与数据进入flink的时间无关)

程序中我们设置的window间隔时间为5s，则窗口划分的结果为：【0-5】【5-10】【10-15】...【50-55】【55-60】，该窗口都是左闭右开区间。

--------------------

那么我们开始在主机s102上输入数据：

![20190531141518585.png][]

控制台输出：

![20190531141611747.png][]

可以看出eventTime为10s，waterMark为:7s(10-3)，所属的window窗口应该是【10-15】，按照我们之前说的，如果想要触发window操作，应该输入一条数据，该数据的eventTime值刚好等于18【waterMark的值为3，使得window结束的时间推迟3s故为：(15+3)】，那我们继续输入数据：

![20190531142204228.png][]

控制台输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70][]

一直输入到16都还没触发window操作，我们继续输入：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 1][]

查看控制台输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 2][]

可以看出当输入eventTime为18的数据时就触发了window操作，window的区间确实是【10-15】，也成功统计出了该范围内的数据。

那我们继续输入数据，看看什么时候触发下一个窗口：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 3][]

控制台输出：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 4][]

看来确实是如果出现一条数据，使得***eventTime=window结束时间+waterMark***即可触发window操作

### 总结一下： ###

水印的目的：处理乱序的数据问题 需要结合window来处理

window触发的条件：  
1、window中必须要数据  
2、waterrMark值=window的结束时间/event-time=window的结束时间+允许乱序的时间(waterrMark值)

--------------------

对于延迟的数据的处理请看下一篇博客：[Flink对于迟到数据的处理][Flink]

视频讲解：[flink水印视频讲解][flink 1]

[https_github.com_18113996630_flink-practice_blob_master_src_main_scala_com_hrong_flink_watermark_WaterMarkFunc01.scala]: https://github.com/18113996630/flink-practice/blob/master/src/main/scala/com/hrong/flink/watermark/WaterMarkFunc01.scala
[flink]: https://www.bilibili.com/video/av87952025
[20190531141518585.png]: /images/20220125/6012e7a154814100971275ba109e28ff.png
[20190531141611747.png]: /images/20220125/bdcefd513ab44887991c8571a97691fb.png
[20190531142204228.png]: /images/20220125/83e964fc615d4429be44e27045eed70a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70]: /images/20220125/47443a0bbc0e42aaa6bd8ab2eb968a36.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 1]: /images/20220125/de2eb8e9bb28456f9e9bee412f192283.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 2]: /images/20220125/37517fae36294b389c97f599c9defae6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 3]: /images/20220125/b90938920ead4d6698305ea6ea574811.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hscDQyMDc_size_16_color_FFFFFF_t_70 4]: /images/20220125/6ae45ac8e93a4e17b3ea5d190ddae44f.png
[Flink]: https://blog.csdn.net/hlp4207/article/details/90717905
[flink 1]: https://www.bilibili.com/video/av54217775