学而不思则罔,思而不学则殆!
The ProcessFunction
ProcessFunction是一个低级的流处理操作，可以访问所有(非循环)流应用程序的基本组件:
Events(流中的事件)statetimers
可以将ProcessFunction看做是具备访问keyed状态和定时器的FlatMapFunction。它通过invoked方法处理从输入流接收到的每个事件。
state
- Process Function可以使用Runtime Context访问 Flink 内部的keyed state , 类似于有状态的函数访问keyed状态。定时器允许应用程序基于处理时间和事件时间响应变化。
timer
- timer允许应用程序对处理时间和事件时间的变化做出反应。每次有事件到达都会调用函数processElement(...),该函数有参数，也就是Context对象，该对象可以访问元素的事件时间戳和TimerService，还有侧输出。
timer 描述TimerService
- TimerService可以用来为将来的事件/处理时间注册回调。当定时器的达到定时时间时，会调用onTimer(...) 方法。 注意：想要访问keyed状态和定时器，则必须在键控流上应用ProcessFunction:
stream.keyBy(...).process(new MyProcessFunction())KeyedProcessFunction 执行流程
- 调用Process方法 传入自定义的KeyedProcessFunction
- 根据类型提取器获取OutPutType
- 创建一个实例化对象并返回 (内部走 Open/processElement/onEventTime/onProcessingTime/ 逻辑)
- 调用transform 方法 @return The transformed {@link DataStream}.
/** {@link org.apache.flink.streaming.api.functions.ProcessFunction} */
/** {@link org.apache.flink.streaming.api.datastream 中的process} */
/**
* Applies the given {@link KeyedProcessFunction} on the input stream, thereby creating a transformed output stream.
*
* <p>The function will be called for every element in the input streams and can produce zero
* or more output elements. Contrary to the {@link DataStream#flatMap(FlatMapFunction)}
* function, this function can also query the time and set timers. When reacting to the firing
* of set timers the function can directly emit elements and/or register yet more timers.
*
* @param keyedProcessFunction The {@link KeyedProcessFunction} that is called for each element in the stream.
*
* @param <R> The type of elements emitted by the {@code KeyedProcessFunction}.
*
* @return The transformed {@link DataStream}.
*/
@PublicEvolving
public <R> SingleOutputStreamOperator<R> process(KeyedProcessFunction<KEY, T, R> keyedProcessFunction) {
TypeInformation<R> outType = TypeExtractor.getUnaryOperatorReturnType(
keyedProcessFunction,
KeyedProcessFunction.class,
1,
2,
TypeExtractor.NO_INDEX,
getType(),
Utils.getCallLocationName(),
true);
// 调用KeyedStream本身的process方法 (每个事件都会调用该方法)
return process(keyedProcessFunction, outType);
}/**
* Applies the given {@link KeyedProcessFunction} on the input stream, thereby creating a transformed output stream.
*
* <p>The function will be called for every element in the input streams and can produce zero
* or more output elements. Contrary to the {@link DataStream#flatMap(FlatMapFunction)}
* function, this function can also query the time and set timers. When reacting to the firing
* of set timers the function can directly emit elements and/or register yet more timers.
*
* @param keyedProcessFunction The {@link KeyedProcessFunction} that is called for each element in the stream.
*
* @param outputType {@link TypeInformation} for the result type of the function.
*
* @param <R> The type of elements emitted by the {@code KeyedProcessFunction}.
*
* @return The transformed {@link DataStream}.
*/
@Internal
public <R> SingleOutputStreamOperator<R> process(
KeyedProcessFunction<KEY, T, R> keyedProcessFunction,
TypeInformation<R> outputType) {
// new
KeyedProcessOperator<KEY, T, R> operator = new KeyedProcessOperator<>(clean(keyedProcessFunction));
return transform("KeyedProcess", outputType, operator);
}查看KeyedProcessOperator内部逻辑
创建成员变量 
初始化上下文对象
每条数据来到都会按此处理KeyedProcessOperator实现了Triggerable
KeyedProcessOperator - invokeUserFunction - on xx Time- 可以看到在
onEventTime或者onProcessingTime方法调用的时候才会调用userFunction.onTimer。那么onEventTime什么时候触发呢？ - .以onEventTime为例
InternalTimerServiceImpl进入到InternalTimerServiceImpl
public void advanceWatermark(long time) throws Exception {
currentWatermark = time;
InternalTimer<K, N> timer;
while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) {
eventTimeTimersQueue.poll();
keyContext.setCurrentKey(timer.getKey());
triggerTarget.onEventTime(timer);
}
}也就是说InternalTimerServiceImpl调用advanceWatermark时我们的onEventTime方法才调用。而advanceWatermark方法的入参time是当前operator的watermark所代表的时间。那么什么时候调用advanceWatermark呢？这个等下再看。 这个方法里面的eventTimeTimersQueue是
/**
* Event time timers that are currently in-flight.
*/
private final KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue;当我们调用时ctx.timerService().registerEventTimeTimer(current.getSystemTimestamp() delay); 就是调用
@Override
public void registerEventTimeTimer(N namespace, long time) {
eventTimeTimersQueue.add(new TimerHeapInternalTimer<>(time, (K) keyContext.getCurrentKey(), namespace));
}向里eventTimeTimersQueue存储TimerHeapInternalTimer（包含key,timestamp等）。 当调用advanceWatermark时，更新currentWatermark，从eventTimeTimersQueue里peek出timer，判断当前watermark的时间是否大于timer里的时间，若大于，则从队列里弹出这个timer调用 triggerTarget.onEventTime(timer) 也就是调用 KeyedProcessOperator.onEventTime，最终调用到里我们自定义OutageFunction的onTimer方法。
// SingleOutputStreamOperator<R>
代码语言：javascript复制 @Override
@PublicEvolving
public <R> SingleOutputStreamOperator<R> transform(String operatorName,
TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {
SingleOutputStreamOperator<R> returnStream = super.transform(operatorName, outTypeInfo, operator);
// inject the key selector and key type
OneInputTransformation<T, R> transform = (OneInputTransformation<T, R>) returnStream.getTransformation();
transform.setStateKeySelector(keySelector);
transform.setStateKeyType(keyType);
return returnStream;
}参考以下大神语 : 注 排名不分前后
- 小C菜鸟
- 岳过山丘
