Process Function (Low-level Operations)

学而不思则罔,思而不学则殆!
The ProcessFunction
ProcessFunction是一个低级的流处理操作，可以访问所有(非循环)流应用程序的基本组件:

• Events(流中的事件)
• state (容错, 一致性，只在Keyed Stream)
• timers (事件时间和处理时间，仅仅适用于keyed Stream)

可以将ProcessFunction看做是具备访问keyed状态和定时器的FlatMapFunction。它通过invoked方法处理从输入流接收到的每个事件。

state

1. Process Function可以使用Runtime Context访问 Flink 内部的keyed state , 类似于有状态的函数访问keyed状态。定时器允许应用程序基于处理时间和事件时间响应变化。

timer

1. timer允许应用程序对处理时间和事件时间的变化做出反应。每次有事件到达都会调用函数processElement(…),该函数有参数，也就是Context对象，该对象可以访问元素的事件时间戳和TimerService，还有侧输出。

TimerService

1. TimerService可以用来为将来的事件/处理时间注册回调。当定时器的达到定时时间时，会调用onTimer(…) 方法。
2. 注意：想要访问keyed状态和定时器，则必须在键控流上应用ProcessFunction:

   stream.keyBy(...).process(new MyProcessFunction())

• KeyedProcessFunction 执行流程

1. 调用Process方法 传入自定义的KeyedProcessFunction
2. 根据类型提取器获取OutPutType
3. 创建一个实例化对象并返回 (内部走 Open/processElement/onEventTime/onProcessingTime/ 逻辑)
4. 调用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) {

        // 每次执行 清空 keyedProcessFunction 疑问 这个clean 方法出现太多次 还不是很清楚
        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

@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;
    }