前言

前面文章Flink中的时间语义 和WaterMark有详细介绍过Flink WaterMark。WaterMark的出现是用来解决乱序时间的处理也就是处理迟到元素的。

WaterMark可以用来平衡计算的完整性和延迟两方面。除非我们选择一种非常保守的水位线策略(最大延时设置的非常大，以至于包含了所有的元素，但结果是非常大的延迟)，否则我们总需要处理迟到的元素。

迟到的元素是指当这个元素来到时，这个元素所对应的窗口已经计算完毕了(也就是说水位线已经没过窗口结束时间了)。这说明迟到这个特性只针对事件时间。

---

处理迟到的元素的策略

DataStream API提供了三种策略来处理迟到元素:

• 直接抛弃迟到的元素
• 将迟到的元素发送到另一条流中去
• 可以更新窗口已经计算完的结果，并发出计算结果。

使用process function抛弃迟到元素

抛弃迟到的元素是event time window operator的默认行为。也就是说一个迟到的元素不会创建一个新的窗口。

process function可以通过比较迟到元素的时间戳和当前水位线的大小来很轻易的过滤掉迟到元素。

使用侧输出(side output)重定向迟到元素

迟到的元素也可以使用侧输出(side output)特性被重定向到另外的一条流中去。迟到元素所组成的侧输出流可以继续处理或者sink到持久化设施中去。

例子

    val readings = env
      .socketTextStream("localhost", 9999, '\n')
      .map(line => {
        val arr = line.split(" ")
        (arr(0), arr(1).toLong * 1000)
      })
      .assignAscendingTimestamps(_._2)

    val countPer10Secs = readings
      .keyBy(_._1)
      .timeWindow(Time.seconds(10))
      .sideOutputLateData(
        new OutputTag[(String, Long)]("late-readings")
      )
      .process(new CountFunction())

    val lateStream = countPer10Secs
      .getSideOutput(
        new OutputTag[(String, Long)]("late-readings")
      )

    lateStream.print()

实现CountFunction:

    class CountFunction extends ProcessWindowFunction[(String, Long),
      String, String, TimeWindow] {
      override def process(key: String,
                           context: Context,
                           elements: Iterable[(String, Long)],
                           out: Collector[String]): Unit = {
        out.collect("窗口共有" + elements.size + "条数据")
      }
    }

下面这个例子展示了ProcessFunction如何过滤掉迟到的元素然后将迟到的元素发送到侧输出流中去。

    val readings: DataStream[SensorReading] = ???
    val filteredReadings: DataStream[SensorReading] = readings
      .process(new LateReadingsFilter)

    // retrieve late readings
    val lateReadings: DataStream[SensorReading] = filteredReadings
      .getSideOutput(new OutputTag[SensorReading]("late-readings"))

    /** A ProcessFunction that filters out late sensor readings and 
      * re-directs them to a side output */
    class LateReadingsFilter 
        extends ProcessFunction[SensorReading, SensorReading] {

      val lateReadingsOut = new OutputTag[SensorReading]("late-readings")

      override def processElement(
          r: SensorReading,
          ctx: ProcessFunction[SensorReading, SensorReading]#Context,
          out: Collector[SensorReading]): Unit = {

        // compare record timestamp with current watermark
        if (r.timestamp < ctx.timerService().currentWatermark()) {
          // this is a late reading => redirect it to the side output
          ctx.output(lateReadingsOut, r)
        } else {
          out.collect(r)
        }
      }
    }

使用allowed lateness迟到元素更新窗口计算结果

由于存在迟到的元素，所以已经计算出的窗口结果是不准确和不完全的。我们可以使用迟到元素更新已经计算完的窗口结果。

如果我们要求一个operator支持重新计算和更新已经发出的结果，就需要在第一次发出结果以后也要保存之前所有的状态。但显然我们不能一直保存所有的状态，肯定会在某一个时间点将状态清空，而一旦状态被清空，结果就再也不能重新计算或者更新了。而迟到的元素只能被抛弃或者发送到侧输出流。

window operator API提供了方法来明确声明我们要等待迟到元素。当使用event-time window，我们可以指定一个时间段叫做allowed lateness。window operator如果设置了allowed lateness，这个window operator在水位线没过窗口结束时间时也将不会删除窗口和窗口中的状态。窗口会在一段时间内(allowed lateness设置的)保留所有的元素。

当迟到元素在allowed lateness时间内到达时，这个迟到元素会被实时处理并发送到触发器(trigger)。当水位线没过了窗口结束时间+allowed lateness时间时，窗口会被删除，并且所有后来的迟到的元素都会被丢弃。

Allowed lateness可以使用allowedLateness()方法来指定，如下所示：

    val readings: DataStream[SensorReading] = ...

    val countPer10Secs: DataStream[(String, Long, Int, String)] = readings
      .keyBy(_.id)
      .timeWindow(Time.seconds(10))
      // process late readings for 5 additional seconds
      .allowedLateness(Time.seconds(5))
      // count readings and update results if late readings arrive
      .process(new UpdatingWindowCountFunction)

      /** A counting WindowProcessFunction that distinguishes between 
      * first results and updates. */
    class UpdatingWindowCountFunction
        extends ProcessWindowFunction[SensorReading,
          (String, Long, Int, String), String, TimeWindow] {

      override def process(
          id: String,
          ctx: Context,
          elements: Iterable[SensorReading],
          out: Collector[(String, Long, Int, String)]): Unit = {

        // count the number of readings
        val cnt = elements.count(_ => true)

        // state to check if this is
        // the first evaluation of the window or not
        val isUpdate = ctx.windowState.getState(
          new ValueStateDescriptor[Boolean](
            "isUpdate",
            Types.of[Boolean]))

        if (!isUpdate.value()) {
          // first evaluation, emit first result
          out.collect((id, ctx.window.getEnd, cnt, "first"))
          isUpdate.update(true)
        } else {
          // not the first evaluation, emit an update
          out.collect((id, ctx.window.getEnd, cnt, "update"))
        }
      }
    }

---

总结

对迟到元素处理，要根据具体业务权衡利弊。
对于不是很重要的数据，并且追求实效性可以直接抛弃。
对于数据实效性可以一定容忍，可以使用WaterMark去延迟处理数据。
对于数据很重要，并且要求很实时计算，可以加入Allowed lateness 不关闭窗口延迟更新。但是注意这会消耗大量的资源。
甚至可以结合WaterMark再加上Allowed lateness来处理延迟数据。