ProcessFunction
是一个低级的流处理操作,可以访问所有(非循环)流应用程序的基本组件:
Events
(流中的事件)state
(容错, 一致性,只在Keyed Stream)timers
(事件时间和处理时间,仅仅适用于keyed Stream)可以将ProcessFunction看做是具备访问keyed状态和定时器的FlatMapFunction。它通过invoked方法处理从输入流接收到的每个事件。
state
timer
TimerService
stream.keyBy(...).process(new MyProcessFunction())
/** {@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
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>
@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;
}
参考以下大神语 : 注 排名不分前后
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。