五、ProcessFunction之类函数相关总结以及SideOutput（侧输出）_processfunction.context.output

一、处理函数是什么？

由代码理解，基本上与process（单词）相关的函数都叫处理函数。

高级的转换流函数，可以实现对数据进行更精确的处理的函数，有些可以获得暴露出的时间戳和水位线。

二、为什么需要使用处理函数？

• 由于之前的转换算子是无法获取时间戳信息和水位线信息以及定时事件等。

三、处理函数如何使用？

1、处理函数的种类：

• ProcessFunction(常用)
• KeyedProcessFunction(常用)
• CoProcessFunction(常用)
• ProcessJoinFunction
• BroadcastProcessFunction
• KeyedBroadcastProcessFunction
• ProcessWindowFunction(常用)
• ProcessAllWindowFunction

特点：

• 都直接或者间接继承了AbstractRichFunction 方法，都可以重写open()和close()方法
• processElement(v: INT, ctx: Context, out: Collector[OUT]), 流中的每一个元素都会调用这个方法，调用结果将会放在Collector数据类型中输出。Context可以访问元素的时间戳，元素的key，以及TimerService时间服务。Context还可以将结果输出到别的流(side outputs)。
• onTimer(timestamp: Long, ctx: OnTimerContext, out: Collector[OUT])是一个回调函数。当之前注册的定时器触发时调用。参数timestamp为定时器所设定的触发的时间戳。Collector为输出结果的集合。OnTimerContext和processElement的Context参数一样，提供了上下文的一些信息，例如firing trigger的时间信息(事件时间或者处理时间)。

1.1、ProcessFunction

如果要使用ProcessFunction，需要该类继承抽象类ProcessFunction()方法并实现其中的抽象方法。
可以不进行Keyby之后使用，并对流中的每条元素都进行遍历以及操作。获取水位线以及定时，侧输出等操作。
1
2

package org.apache.flink.streaming.api.functions;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import org.apache.flink.streaming.api.TimeDomain;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

@PublicEvolving
public abstract class ProcessFunction<I, O> extends AbstractRichFunction {
   
	private static final long serialVersionUID = 1L;
	public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;
	public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {
   }

	
	public abstract class Context {
   
		public abstract Long timestamp();
		public abstract TimerService timerService();
		public abstract <X> void output(OutputTag<X> outputTag, X value);
	}

	public abstract class OnTimerContext extends Context {
   
		public abstract TimeDomain timeDomain();
	}

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

特点：

1. 前面可以没有keyBy()算子。
2. 可以对一条流中每条数据进行遍历操作。
3. 可以获取操作时间，水位线。
4. 可以进行注册定时器操作。
5. 可以向下游发送数据。

1.2、CoProcessFunction

源码如下：

package org.apache.flink.streaming.api.functions.co;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.functions.AbstractRichFunction;
import org
1
2
3
4
5