ProcessFunction API（底层 API）

我们之前学习的转换算子是无法访问事件的时间戳信息和水位线信息的。而这在一些应用场景下，极为重要。例如 MapFunction 这样的 map 转换算子就无法访问时间戳或者当前事件的事件时间。

基于此，DataStream API 提供了一系列的 Low-Level 转换算子。可以访问时间戳、水位线以及注册定时事件。还可以输出特定的一些事件，例如超时事件等。Process Function 用来构建事件驱动的应用以及实现自定义的业务逻辑 (使用之前的 window 函数和转换算子无法实现)。例如，Flink-SQL 就是使用 Process Function 实现的。

Flink 提供了 8 个 Process Function：

• ProcessFunction
• KeyedProcessFunction
• CoProcessFunction
• ProcessJoinFunction
• BroadcastProcessFunction
• KeyedBroadcastProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

KeyedProcessFunction

KeyedProcessFunction 用来操作 KeyedStream。KeyedProcessFunction 会处理流的每一个元素，输出为 0 个、1 个或者多个元素。所有的 Process Function 都继承自 RichFunction 接口，所以都有 open() 、 close() 和 getRuntimeContext() 等方法。而 KeyedProcessFunction [KEY, IN, OUT] 还额外提供了两个方法:

• processElement (v: IN, ctx: Context, out: Collector [OUT]), 流中的每一个元素都会调用这个方法，调用结果将会放在 Collector 数据类型中输出。Context 可以访问元素的时间戳，元素的 key，以及 TimerService 时间服务。Context 还可以将结果输出到别的流 (side outputs)。
• onTimer (timestamp: Long, ctx: OnTimerContext, out: Collector [OUT]) 是一个回调函数。当之前注册的定时器触发时调用。参数 timestamp 为定时器所设定的触发的时间戳。Collector 为输出结果的集合。OnTimerContext 和 processElement 的 Context 参数一样，提供了上下文的一些信息，例如 firing trigger 的时间信息 (事件时间或者处理时间)。

下面用水位线传播规则来示例

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val stream1 = env
      .socketTextStream("localhost", 9999, '\n')
      .map(line => {
        val arr = line.split(" ")
        (arr(0), arr(1).toLong * 1000L)
      })
      // 没有设置最大延迟时间 默认为Long的最小值
      .assignAscendingTimestamps(_._2)

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

    stream1
      .union(stream2)
      .keyBy(_._1)
      .process(new Keyed)
      .print()

    env.execute()

  }

  class Keyed extends KeyedProcessFunction[String, (String, Long), String] {
    // 每到一条数据就会调用一次
    override def processElement(i: (String, Long), context: KeyedProcessFunction[String, (String, Long), String]#Context, collector: Collector[String]): Unit = {
      // 输出当前的水位线
      collector.collect("当前的水位线是：" + context.timerService().currentWatermark())
    }
  }

启动第一个 nc

nc -lk 9999
a 1 #等待水位线结束后再输入第二个nc的 a 100
a 2 #等待水位线结束后在输入第一个nc的 a 3
a 3

启动第二个 nc

nc -lk 9998
a 100 #等待水位线结束后在输入第二个nc的 a 101
a 101 #等待水位线结束后在输入第一个nc的 a 2

输出结果

当前的水位线是： -9223372036854775808 #当前水位线存在两个long的最小值
当前的水位线是： -9223372036854775808 #当前水位线存在 1*1000ms-1ms=999ms 和 一个long的最小值
当前的水位线是： 999 # 当前水位线存在 101*1000ms-1ms=100999ms 和 999ms
当前的水位线是： 999 # 当前水位线存在 101*1000ms-1ms=100999ms 和 2*1000ms-1ms=1999ms
当前的水位线是： 1999 # 当前水位线存在 101*1000ms-1ms=100999ms 和 2*1000ms-1ms=2999ms

由此可以看出我们水位线的传播规则是输出当前水位线中的最小值，然后不断更新。

TimerService 和 定时器（Timers）

Context 和 OnTimerContext 所持有的 TimerService 对象拥有以下方法:

• currentProcessingTime(): Long 返回当前处理时间
• currentWatermark(): Long 返回当前水位线的时间戳
• registerProcessingTimeTimer(timestamp: Long): Unit 会注册当前 key 的 processing time 的 timer。当 processing time 到达定时时间时，触发 timer。
• registerEventTimeTimer(timestamp: Long): Unit 会注册当前 key 的 event time timer。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。
• deleteProcessingTimeTimer(timestamp: Long): Unit 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。
• deleteEventTimeTimer(timestamp: Long): Unit 删除之前注册的事件时间定时器，如果没有此时间戳的定时器，则不执行。

当定时器 timer 触发时，执行回调函数 onTimer() 。 processElement() 方法和 onTimer() 方法是同步（不是异步）方法，这样可以避免并发访问和操作状态。

注意：定时器 timer 只能在 keyed streams 上面使用。

针对每一个 key 和 timestamp，只能注册一个定期器。也就是说，每一个 key 可以注册多个定时器，但在每一个时间戳只能注册一个定时器。KeyedProcessFunction 默认将所有定时器的时间戳放在一个优先队列中。在 Flink 做检查点操作时，定时器也会被保存到状态后端中。

我们来看看基于事件时间的简单使用示例

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val stream = env
      .socketTextStream("localhost", 9999, '\n')
      .map(line => {
        val arr = line.split(" ")
        (arr(0), arr(1).toLong * 1000L)
      })
      .assignAscendingTimestamps(_._2)
      .keyBy(_._1)
      .process(new Keyed)
      .print()
    env.execute()
  }

  class Keyed extends KeyedProcessFunction[String, (String, Long), String] {
    override def processElement(i: (String, Long), context: KeyedProcessFunction[String, (String, Long), String]#Context, collector: Collector[String]): Unit = {
      // 注册一个定时器：事件携带的时间戳+10s
      context.timerService().registerEventTimeTimer(i._2 + 10 * 1000L)
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, (String, Long), String]#OnTimerContext, out: Collector[String]): Unit = {
      out.collect("定时器触发了！" + "定时器执行的时间戳是：" + timestamp)
    }
  }

启动 nc

nc -lk 9999
a 1
a 12
a 23
a 34

输出结果

定时器触发了！定时器执行的时间戳是：11000
定时器触发了！定时器执行的时间戳是：22000
定时器触发了！定时器执行的时间戳是：33000

我们再来看看机器时间的示例

  def main(args: Array[String]): Unit = {
    def main(args: Array[String]): Unit = {
      val env = StreamExecutionEnvironment.getExecutionEnvironment
      env.setParallelism(1)

      val stream = env
        .socketTextStream("localhost", 9999, '\n')
        .map(line => {
          val arr = line.split(" ")
          (arr(0), arr(1).toLong * 1000L)
        })
        .assignAscendingTimestamps(_._2)
        .keyBy(_._1)
        .process(new Keyed)
        .print()
      env.execute()
    }

    class Keyed extends KeyedProcessFunction[String, (String, Long), String] {
      override def processElement(i: (String, Long), context: KeyedProcessFunction[String, (String, Long), String]#Context, collector: Collector[String]): Unit = {
        // 注册一个定时器：当前机器时间+10s
        context.timerService().registerProcessingTimeTimer(context.timerService().currentProcessingTime() + 10 * 1000L)
      }

      override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, (String, Long), String]#OnTimerContext, out: Collector[String]): Unit = {
        out.collect("定时器触发了！" + "定时器执行的时间戳是：" + timestamp)
      }
    }

启动 nc

nc -lk 9999
a 166666

输出结果

定时器触发了！定时器执行的时间戳是：1671801578000 # 当前机器时间 + 10s，与事件的时间无关