【Flink】如何处理数据延迟

601-赵同学

发表文章数:191

首页 » 大数据 » 正文


迟到数据

概述

官网给出的定义:Late elements are elements that arrive after the system’s event time clock (as signaled by the watermarks) has already passed the time of the late element’s timestamp. (迟到数据是指系统的事件时间时钟(由水印指示)在经过延迟元素时间戳之后的时间到达的元素。)

以下观点均是看完罗西的思考在博客园发的一篇文章——《[白话解析] Flink的Watermark机制》的总结

所以迟到数据可以说是一种特殊的乱序数据,它没有被watermark和Window机制处理,因为是在窗口关闭后才到达的数据。一般这种情况有三种处理办法
  1.重新激活已经关闭的窗口并重新计算以修正结果。
  2.将迟到数据收集起来另外处理。
  3.将迟到数据视为错误消息并丢弃。

 
Flink默认采用第三种方法,将迟到数据视为错误消息丢弃。想要使用前两种方法需要使用到sideOutput机制allowedLateness机制
sideOutput机制可以将迟到事件单独放入一个数据流分支,这会作为 window 计算结果的副产品,以便用户获取并对其进行特殊处理。
allowedLateness机制允许用户设置一个允许的最大迟到时长。Flink 会在窗口关闭后一直保存窗口的状态直至超过允许迟到时长,这期间的迟到事件不会被丢弃,而是默认会触发窗口重新计算。
所以,如果要设置允许延迟的时间,可以通过DataStream.allowedLateness(lateness: Time)。如果要保存延迟数据要通过sideOutputLateData(outputTag: OutputTag[T])来保存。而要获取已经保存的延迟数据,则要通过DataStream.getSideOutput(tag: OutputTag[X])


实例

import org.apache.commons.lang.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, OutputTag, StreamExecutionEnvironment, WindowedStream}
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 HandleLatenessDemo {
  //定义一个对象,将时间戳转换为时间
  val sdf = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss:SSS")

  def main(args: Array[String]): Unit = {
    /**
     * 1、监听某主机的9999端口,读取socket数据(格式为  name:timestamp)
     * 2、给当前进入flink程序的数据加上waterMark,值为eventTime-3s
     * 3、根据name值进行分组,根据窗口大小为5s划分窗口,设置允许迟到时间为2s,依次统计窗口中各name值的数据
     * 4、输出统计结果以及迟到数据
     * 5、启动Job
     */
    // 1.获取执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 1.1 将event time设置为流数据时间类型
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 2.获取socket数据源
    // 2.1 socketTextStream可以有hostname/ port/ delimiter/ maxRetry/ 四个参数,这里设置一下前三个,方便后续修改
    val hostname = "node01"
    val port = 9999
    val delimiter = '/n'

    val socketStream: DataStream[String] = env.socketTextStream(hostname, port, delimiter)

    // 3.将输入的数据进行转换
    import org.apache.flink.api.scala._
    val data: DataStream[(String, Long)] = socketStream.map(line => {
      // 比如,输入的数据格式是:name 时间戳
      try {
        val item: Array[String] = line.split(",")
        (item(0).trim, item(1).trim.toLong)
      } catch {
        case _: Exception => println("输入的数据格式不合法")
          ("spark", 0L)
      }
    }).filter(data => data._1.equals("0") && data._2 != 0L)

    // 4.对数据流中的元素分配时间戳,并定期地创建水印(20ms),监控时间的进度
    val watermarkDataStream: DataStream[(String, Long)] = data.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[(String, Long)] {
      // 当前最大时间戳
      private var currentMaxTimestamp = 0L
      // 最大延迟时间
      val maxOutOfOrderness = 3000L
      var lastEmittedWatermark = Long.MinValue

      /**
       * 获取当前水印
       * 该方法会被周期性地执行,20ms一个周期
       */
      override def getCurrentWatermark: Watermark = {
        // 允许最大延迟 3s
        val potentialWatermark: Long = currentMaxTimestamp - maxOutOfOrderness
        new Watermark(potentialWatermark)
      }

      override def extractTimestamp(element: (String, Long), previousElementTimestamp: Long): Long = {
        //将元素的时间字段作为该数据的timestamp
        val time: Long = element._2
        if (time > currentMaxTimestamp) {
          currentMaxTimestamp = time
        }
        val outData: String = String.format("key: %s EventTime: %s watermark: %s", element._1, sdf.format(time), sdf.format(getCurrentWatermark.getTimestamp))
        println(outData)
        time
      }
    })

    // 5.定义一个侧输出流
    val lateData: OutputTag[(String, Long)] = new OutputTag[(String, Long)]("late")

    // 6.对水印数据聚合并引入窗口
    val windowedStream: WindowedStream[(String, Long), Tuple, TimeWindow] = watermarkDataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5L)))

    // 7.获取带有延迟时间的数据流
    val result: DataStream[String] = windowedStream.allowedLateness(Time.seconds(2)) //允许迟到2s
      .sideOutputLateData(lateData)
      .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] = ArrayBuffer[String]()
          val iterator: Iterator[(String, Long)] = input.iterator
          // 将所有的事件时间放到集合中
          while (iterator.hasNext) {
            val tuple: (String, Long) = iterator.next()
            timeArr.append(sdf.format(tuple._2))
          }
          val outData = String.format("key: %s   data: %s   windowStartTime: %s    windowEndTime: %s",
            key.toString,
            timeArr.mkString(","),
            sdf.format(window.getStart),
            sdf.format(window.getEnd)
          )
          out.collect(outData)
        }
      })
    result.print("Window的计算结果>>>")

    val late: DataStream[(String, Long)] = result.getSideOutput(lateData)
    late.print("迟到的数据>>>")

    // 启动执行环境
    env.execute("HandleLatenessDemo")
  }
}

未经允许不得转载:作者:601-赵同学, 转载或复制请以 超链接形式 并注明出处 拜师资源博客
原文地址:《【Flink】如何处理数据延迟》 发布于2020-05-11

分享到:
赞(0) 打赏

评论 抢沙发

评论前必须登录!

  注册



长按图片转发给朋友

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

Vieu3.3主题
专业打造轻量级个人企业风格博客主题!专注于前端开发,全站响应式布局自适应模板。

登录

忘记密码 ?

您也可以使用第三方帐号快捷登录

Q Q 登 录
微 博 登 录