描述
- 数据在KeyBy后对key按hash结果进入对应的KeyedProcessFunction subtask
- 同个key的数据的所有数据都会进入同一个KeyedProcessFunction subtask
- 不同key的数据也有可能进入同一个KeyedProcessFunction subtask
- 同一个KeyedProcessFunction subtask中所有数据共享普通变量
- 同一个KeyedProcessFunction subtask中同一个key的数据共享状态变量，不同key的数据不共享状态变量
- 不同KeyedProcessFunction subtask中所有变量均不共享
- 提供了RuntimeContext的使用
- 提供了Watermark和ProcessingTime的访问
- 提供了timerService的使用，当数据即将触发定时器时，先执行processElement函数，再执行onTimer函数
- 提供了侧输出流的使用
输入
KeyedStream
输出
DataStream
KeyedProcessFunction
声明一个自定义KeyedProcessFunction类
class MyProcessFunction(自定义类参数) extends KeyedProcessFunction[key数据类型, input数据类型, output数据类型] {
// 必须实现processElement方法
override def processElement(value: input数据类型, ctx: KeyedProcessFunction[key数据类型, input数据类型, output数据类型]#Context, out: Collector[output数据类型]): Unit = {
...
}
}
// 使用
dataStream
.keyBy(...)
.process(new MyProcessFunction(...))
// 在processFunction中使用状态一个valueState
lazy val myState: ValueState[Long] = getRuntimeContext.getState(
new ValueStateDescriptor[Long]("myState", classOf[Long])
)
// 在processFunction中使用timerService计时器功能
// 声明一个基于eventTime的计时器, 当该task中的数据eventTime到达触发时间戳时，就会调用onTimer方法，并清除该计时器。 ctx在processElement方法和onTimer方法中均能使用
ctx.timerService().registerEventTimeTimer(触发时间戳，单位毫秒)
// 声明一个基于processTime的计时器，当processTime到达触发时间戳时，该task会调用onTimer方法，并清除计时器
ctx.timerService().registerProcessingTimeTimer(触发时间戳，单位毫秒)
// 手动删除一个eventTime计时器，需要指定计时器对应的时间戳
ctx.timerService().deleteEventTimeTimer(计时器触发时间戳)
// 手动删除一个processTime计时器，需要指定计时器对应的时间戳
ctx.timerService().deleteProcessingTimeTimer(计时器触发时间戳)
// 实现onTimer方法
override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[key数据类型, input数据类型, output数据类型]#OnTimerContext, out: Collector[output数据类型]): Unit = {
...
//其中timestamp为声明计时器时指定的eventTime或processTime
}
// 访问task中的watermark，watermark与eventTime对齐
ctx.timerService().currentWatermark()
// 访问task中的processTime
ctx.timerService().currentProcessingTime()
// 使用侧输出流
ctx.output(new OutputTag[output数据类型]("定义测输出流id"), output value)
