1.简介

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

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

Flink 提供了8 个Process Function：

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

2.算子的介绍

2.1KeyedProcessFunction

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

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

实例：监控温度传感器的温度值，如果温度值在一秒钟之内(processing time)连续上升，则报警。

object ProcessFunctionTest {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)// 读取数据val inputStream = env.socketTextStream("localhost", 7777)// 先转换成样例类类型（简单转换操作）val dataStream = inputStream.map( data => {val arr = data.split(",")SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)} )
//      .keyBy(_.id)
//      .process( new MyKeyedProcessFunction )val warningStream = dataStream.keyBy(_.id).process( new TempIncreWarning(1000L) )warningStream.print()env.execute("process function test")}
}// 实现自定义的KeyedProcessFunction
class TempIncreWarning(interval: Long) extends KeyedProcessFunction[String, SensorReading, String]{// 定义状态：保存上一个温度值进行比较，保存注册定时器的时间戳用于删除// 懒加载；// 状态变量会在检查点操作时进行持久化，例如hdfs// 只会初始化一次，单例模式// 在当机重启程序时，首先去持久化设备寻找名为`last-temp`的状态变量，如果存在，则直接读取。不存在，则初始化。// 用来保存最近一次温度// 默认值是0.0lazy val lastTempState: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("last-temp", classOf[Double]))lazy val timerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("timer-ts", classOf[Long]))override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {// 先取出状态val lastTemp = lastTempState.value()val timerTs = timerTsState.value()// 更新温度值lastTempState.update(value.temperature)// 当前温度值和上次温度进行比较if( value.temperature > lastTemp && timerTs == 0 ){// 如果温度上升，且没有定时器，那么注册当前时间10s之后的定时器// currentProcessingTime(): Long 返回当前处理时间val ts = ctx.timerService().currentProcessingTime() + interval// registerProcessingTimeTimer 会注册当前key的processing time的timer。当processing time到达定时时间时，触发timer。ctx.timerService().registerProcessingTimeTimer(ts)timerTsState.update(ts)} else if( value.temperature < lastTemp ){// 如果温度下降，那么删除定时器 删除之前注册处理时间定时器。如果没有这个时间戳的定时器，则不执行。ctx.timerService().deleteProcessingTimeTimer(timerTs)timerTsState.clear()}}override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {out.collect("传感器" + ctx.getCurrentKey + "的温度连续" + interval/1000 + "秒连续上升")timerTsState.clear()}
}// KeyedProcessFunction功能测试MyRichMapperMyRichMapperMyRichMapper
class MyKeyedProcessFunction extends KeyedProcessFunction[String, SensorReading, String]{var myState: ValueState[Int] = _override def open(parameters: Configuration): Unit = {myState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("mystate", classOf[Int]))}override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {ctx.getCurrentKeyctx.timestamp()// 返回当前水位线的时间戳ctx.timerService().currentWatermark()// 会注册当前key的event time timer。当水位线大于等于定时器注册的时间时，触发定时器执行回调函数。ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 60000L)}override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {}
}

这里对一些函数中的方法进行介绍：

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

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

当定时器timer 触发时，会执行回调函数onTimer()。注意定时器timer 只能在keyed streams 上面使用。

2.2侧输出流（SideOutput）

大部分的DataStream API 的算子的输出是单一输出，也就是某种数据类型的流。除了split 算子，可以将一条流分成多条流，这些流的数据类型也都相同。processfunction 的side outputs 功能可以产生多条流，并且这些流的数据类型可以不一样。一个side output 可以定义为OutputTag[X]对象，X 是输出流的数据类型。processfunction 可以通过Context 对象发射一个事件到一个或者多个side outputs。

object SideOutputTest {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)// 读取数据val inputStream = env.socketTextStream("localhost", 7777)// 先转换成样例类类型（简单转换操作）val dataStream = inputStream.map( data => {val arr = data.split(",")SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)} )val highTempStream = dataStream.process( new SplitTempProcessor(30.0) )highTempStream.print("high")highTempStream.getSideOutput(new OutputTag[(String, Long, Double)]("low")).print("low")env.execute("side output test")}
}// 实现自定义ProcessFunction，进行分流
class SplitTempProcessor(threshold: Double) extends ProcessFunction[SensorReading, SensorReading]{override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, SensorReading]#Context, out: Collector[SensorReading]): Unit = {if( value.temperature > threshold ){// 如果当前温度值大于30，那么输出到主流out.collect(value)} else {// 如果不超过30度，那么输出到侧输出流ctx.output(new OutputTag[(String, Long, Double)]("low"), (value.id, value.timestamp, value.temperature))}}
}

2.3CoProcessFunction

对于两条输入流，DataStream API 提供了CoProcessFunction 这样的low-level操作。CoProcessFunction 提供了操作每一个输入流的方法: processElement1()和processElement2()。

类似于ProcessFunction，这两种方法都通过Context 对象来调用。这个Context对象可以访问事件数据，定时器时间戳，TimerService，以及side outputs。CoProcessFunction 也提供了onTimer()回调函数。

case class OrderEvent(orderId: Long, eventType: String, txId: String, timestamp: Long)
case class ReceiptEvent(txId: String, payChannel: String, timestamp: Long)object TxMatch {def main(args: Array[String]): Unit = {val env = StreamExecutionEnvironment.getExecutionEnvironmentenv.setParallelism(1)env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)// 1. 读取订单事件数据val resource1 = getClass.getResource("/OrderLog.csv")val orderEventStream = env.readTextFile(resource1.getPath)//    val orderEventStream = env.socketTextStream("localhost", 7777).map( data => {val arr = data.split(",")OrderEvent(arr(0).toLong, arr(1), arr(2), arr(3).toLong)} ).assignAscendingTimestamps(_.timestamp * 1000L).filter(_.eventType == "pay").keyBy(_.txId)// 2. 读取到账事件数据val resource2 = getClass.getResource("/ReceiptLog.csv")val receiptEventStream = env.readTextFile(resource2.getPath)//    val orderEventStream = env.socketTextStream("localhost", 7777).map( data => {val arr = data.split(",")ReceiptEvent(arr(0), arr(1), arr(2).toLong)} ).assignAscendingTimestamps(_.timestamp * 1000L).keyBy(_.txId)// 3. 合并两条流，进行处理val resultStream = orderEventStream.connect(receiptEventStream).process( new TxPayMatchResult() )resultStream.print("matched")resultStream.getSideOutput(new OutputTag[OrderEvent]("unmatched-pay")).print("unmatched pays")resultStream.getSideOutput(new OutputTag[ReceiptEvent]("unmatched-receipt")).print("unmatched receipts")env.execute("tx match job")}
}class TxPayMatchResult() extends CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]{// 定义状态，保存当前交易对应的订单支付事件和到账事件lazy val payEventState: ValueState[OrderEvent] = getRuntimeContext.getState(new ValueStateDescriptor[OrderEvent]("pay", classOf[OrderEvent]))lazy val receiptEventState: ValueState[ReceiptEvent] = getRuntimeContext.getState(new ValueStateDescriptor[ReceiptEvent]("receipt", classOf[ReceiptEvent]))// 侧输出流标签val unmatchedPayEventOutputTag = new OutputTag[OrderEvent]("unmatched-pay")val unmatchedReceiptEventOutputTag = new OutputTag[ReceiptEvent]("unmatched-receipt")override def processElement1(pay: OrderEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {// 订单支付来了，要判断之前是否有到账事件val receipt = receiptEventState.value()if( receipt != null ){// 如果已经有receipt，正常输出匹配，清空状态out.collect((pay, receipt))receiptEventState.clear()payEventState.clear()} else{// 如果还没来，注册定时器开始等待5秒ctx.timerService().registerEventTimeTimer(pay.timestamp * 1000L + 5000L)// 更新状态payEventState.update(pay)}}override def processElement2(receipt: ReceiptEvent, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#Context, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {// 到账事件来了，要判断之前是否有pay事件val pay = payEventState.value()if( pay != null ){// 如果已经有pay，正常输出匹配，清空状态out.collect((pay, receipt))receiptEventState.clear()payEventState.clear()} else{// 如果还没来，注册定时器开始等待3秒ctx.timerService().registerEventTimeTimer(receipt.timestamp * 1000L + 3000L)// 更新状态receiptEventState.update(receipt)}}override def onTimer(timestamp: Long, ctx: CoProcessFunction[OrderEvent, ReceiptEvent, (OrderEvent, ReceiptEvent)]#OnTimerContext, out: Collector[(OrderEvent, ReceiptEvent)]): Unit = {// 定时器触发，判断状态中哪个还存在，就代表另一个没来，输出到侧输出流if( payEventState.value() != null ){ctx.output(unmatchedPayEventOutputTag, payEventState.value())}if( receiptEventState.value() != null ){ctx.output(unmatchedReceiptEventOutputTag, receiptEventState.value())}// 清空状态receiptEventState.clear()payEventState.clear()}
}