在 Flink 的多层 API中，处理函数是最底层的API，是所有转换算子的一个概括性的表达，可以自定义处理逻辑
在处理函数中，我们直面的就是数据流中最基本的元素：数据事件（event）、状态（state）以及时间（time）。这就相当于对流有了完全的控制权
基本处理函数主要是定义数据流的转换操作，其所对应的函数类为ProcessFunction
处理函数的功能和使用
对于常用的转换算子来说：
- MapFunction只能获取到当前的数据；
- AggregateFunction 中除数据外，还可以获取到当前的状态（以累加器 Accumulator 形式出现）；
- RichMapFunction提供了获取运行时上下文的方法 getRuntimeContext()；
但是无论那种算子，如果我们想要访问事件的时间戳，或者当前的水位线信息，都是完全做不到的
与时间相关的操作只能用时间窗口去处理，但如果要求对时间有更精细的控制，需要能够获取水位线，甚至要“把控时间”、定义什么时候做什么事，这就不是基本的时间窗口能够实现的了
因此需要使用处理函数：
- 处理函数提供了一个“定时服务”（TimerService），我们可以通过它访问流中的事件（event）、时间戳（timestamp）、水位线（watermark），甚至可以注册“定时事件”
- 处理函数继承了
AbstractRichFunction抽象类，所以拥有富函数类的所有特性，同样可以访问状态（state）和其他运行时信息 - 处理函数还可以直接将数据输出到侧输出流（side output）中
处理函数的简单使用：基于 DataStream 调用.process()方法就；方法需要传入一个 ProcessFunction 作为参数，用来定义处理逻辑：
stream.process(new MyProcessFunction())简单示例：
public class ProcessFunctionTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env
.addSource(new ClickSource())
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Event>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
@Override
public long extractTimestamp(Event event, long l) {
return event.timestamp;
}
})
)
.process(new ProcessFunction<Event, String>() {
@Override
public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
if (value.user.equals("Mary")) {
out.collect(value.user);
} else if (value.user.equals("Bob")) {
out.collect(value.user);
out.collect(value.user);
}
System.out.println(ctx.timerService().currentWatermark());
}
})
.print();
env.execute();
}
}在 ProcessFunction 中重写了.processElement()方法（参数：输入，上下文对象，输出），自定义处理逻辑
ProcessFunction 解析
源码解析
源码如下：
public abstract class ProcessFunction<I, O> extends AbstractRichFunction {
private static final long serialVersionUID = 1L;
/**
* Process one element from the input stream.
*
* <p>This function can output zero or more elements using the {@link Collector} parameter and
* also update internal state or set timers using the {@link Context} parameter.
*
* @param value The input value.
* @param ctx A {@link Context} that allows querying the timestamp of the element and getting a
* {@link TimerService} for registering timers and querying the time. The context is only
* valid during the invocation of this method, do not store it.
* @param out The collector for returning result values.
* @throws Exception This method may throw exceptions. Throwing an exception will cause the
* operation to fail and may trigger recovery.
*/
public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;
/**
* Called when a timer set using {@link TimerService} fires.
*
* @param timestamp The timestamp of the firing timer.
* @param ctx An {@link OnTimerContext} that allows querying the timestamp of the firing timer,
* querying the {@link TimeDomain} of the firing timer and getting a {@link TimerService}
* for registering timers and querying the time. The context is only valid during the
* invocation of this method, do not store it.
* @param out The collector for returning result values.
* @throws Exception This method may throw exceptions. Throwing an exception will cause the
* operation to fail and may trigger recovery.
*/
public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {}
/**
* Information available in an invocation of {@link #processElement(Object, Context, Collector)}
* or {@link #onTimer(long, OnTimerContext, Collector)}.
*/
public abstract class Context {
/**
* Timestamp of the element currently being processed or timestamp of a firing timer.
*
* <p>This might be {@code null}, for example if the time characteristic of your program is
* set to {@link org.apache.flink.streaming.api.TimeCharacteristic#ProcessingTime}.
*/
public abstract Long timestamp();
/** A {@link TimerService} for querying time and registering timers. */
public abstract TimerService timerService();
/**
* Emits a record to the side output identified by the {@link OutputTag}.
*
* @param outputTag the {@code OutputTag} that identifies the side output to emit to.
* @param value The record to emit.
*/
public abstract <X> void output(OutputTag<X> outputTag, X value);
}
/**
* Information available in an invocation of {@link #onTimer(long, OnTimerContext, Collector)}.
*/
public abstract class OnTimerContext extends Context {
/** The {@link TimeDomain} of the firing timer. */
public abstract TimeDomain timeDomain();
}
}
可以看到抽象类 ProcessFunction 继承了 AbstractRichFunction，有两个泛型类型参数：
I 表示 Input，也就是输入的数据类型；O 表示 Output，也就是处理完成之后输出的数据类型
其内部单独定义了两个方法：一个是必须要实现的抽象方法.processElement()；另一个是非抽象方法.onTimer()
.processElement()：用于“处理元素”，定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次，参数包括三个：输入数据值 value，上下文 ctx，以及“收集器”（Collector）out。方法没有返回值，处理之后的输出数据是通过收集器 out 来定义- value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类型一致
- ctx：类型是
ProcessFunction中定义的内部抽象类Context，表示当前运行的上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服务”(TimerService)，以及可以将数据发送到“侧输出流”（side output）的方法.output() - out：“收集器”（类型为 Collector），用于返回输出数据。使用方式与 flatMap算子中的收集器完全一样，直接调用
out.collect()方法就可以向下游发出一个数据。这个方法可以多次调用，也可以不调用
.onTimer()：用于定义定时触发的操作；这个方法只有在注册好的定时器触发的时候才会调用（在 Flink 中，只有“按键分区流”KeyedStream才支持设置定时器的操作），而定时器是通过“定时服务”TimerService来注册的- 参数：时间戳（timestamp），上下文（ctx），收集器（out）【这里的时间戳是指设置好的触发时间，在事件时间语义下就是水位线】
利用onTimer，可以自定义数据按照时间分组、定时触发计算输出结果，这样就实现了窗口的功能
处理函数分类
- ProcessFunction：最基本的处理函数，基于
DataStream直接调用.process()时作为参数传入 - KeyedProcessFunction：对流按键分区后的处理函数，基于
KeyedStream调用.process()时作为参数传入（要想使用定时器必须基于 KeyedStream ） - ProcessWindowFunction：开窗之后的处理函数，也是全窗口函数的代表。基于
WindowedStream调用.process()时作为参数传入 - ProcessAllWindowFunction：开窗之后的处理函数，基于
AllWindowedStream调用.process()时作为参数传入 - CoProcessFunction：合并（connect）两条流之后的处理函数，基于
ConnectedStreams调用.process()时作为参数传入 - ProcessJoinFunction：间隔连接（interval join）两条流之后的处理函数，基于
IntervalJoined调用.process()时作为参数传入 - BroadcastProcessFunction：广播连接流处理函数，基于
BroadcastConnectedStream调用.process()时作为参数传入（这里的“广播连接流”BroadcastConnectedStream，是一个未 keyBy 的普通 DataStream 与一个广播流（BroadcastStream）做连接（conncet）之后的产物） - KeyedBroadcastProcessFunction：按键分区的广播连接流处理函数，同样是基于
BroadcastConnectedStream调用.process()时作为参数传入（这时的广播连接流，是一个KeyedStream与广播流（BroadcastStream）做连接之后的产物）
学习课程链接：【尚硅谷】Flink1.13实战教程（涵盖所有flink-Java知识点）_哔哩哔哩_bilibili
