Process Function (Low-level Operations)

原创

用户7283316

修改于 2020-05-08 14:19:20

7390

修改于 2020-05-08 14:19:20

文章被收录于专栏：F

学而不思则罔,思而不学则殆!

The ProcessFunction

ProcessFunction是一个低级的流处理操作，可以访问所有(非循环)流应用程序的基本组件:

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

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

state

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

timer

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

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

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 删除。

css