1. 处理函数介绍
流处理API,无论是基本的转换
、聚合
、还是复杂的窗口
操作,都是基于DataStream
进行转换的,所以统称为DataStreamAPI
,这是Flink编程的核心。
但其实Flink为了更强大的表现力和易用性,Flink本身提供了多层API,DataStreamAPI
只是中间一环,如下图所示:
在更底层,Flink可以不定义任何具体的算子(比如 map
,filter
,或者 window
),而只是提炼出一个统一的“处理”(process)
操作——它是所有转换算子的一个概括性的表达,可以自定义处理逻辑,所以这一层接口就被叫作“处理函数”(process function)
。
在处理函数中,操作的就是数据流中最基本的元素:数据事件(event)
、状态(state
)以及时间(time)
。这就相当于对流有了完全的控制权。处理函数比较抽象,没有具体的操作,所以对于一些常见的简单应用(比如求和、开窗口)会显得有些麻烦;不过正是因为它不限定具体做什么,所以理论上我们可以做任何事情,实现所有需求。
2. 处理函数的分类
DataStream
在调用一些转换方法之后,有可能生成新的流类型;例如调用.keyBy()
之后得到 KeyedStream
,进而再调用.window()
之后得到 WindowedStream
。对于不同类型的流
,其实都可以直接调用.process()
方法进行自定义处理,这时传入的参数就都叫作处理函数
。当然,它们尽管本质相同,都是可以访问状态和时间信息的底层 API,可彼此之间也会有所差异。
Flink 提供了 8 个不同的处理函数:
- 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()
时作为参数传入。与BroadcastProcessFunction
不同的是,这时的广播连接流,是一个KeyedStream
与广播流(BroadcastStream)
做连接之后的产物。
2.1 基本处理函数(ProcessFunction)
处理函数主要是定义数据流的转换操作,所以也可以把它归到转换算子中。在Flink 中几乎所有转换算子都提供了对应的函数类接口,处理函数也不例外;它所对应的函数类,就叫作 ProcessFunction
。
2.1.1 处理函数的功能和使用
转换算子一般只是针对某种具体操作来定义的,能够拿到的信息比较有限。比如Map
算子只能获取当前的数据;而想窗口聚合复杂的操作AggregateFunction
中除数据外,还可以获取到当前的状态(以累加器 Accumulator 形式出现)
。另外还有富函数类
,比如 RichMapFunction
,它提供了获取运行时上下文的方法 getRuntimeContext()
,可以拿到状态
,还有并行度
、任务名称
之类的运行时信息。
但无论那种算子,如果想要访问事件的时间戳
,或者当前的水位线信息
,都是获取不到的。但是处理函数可以获取,处理函数提供了一个“定时服务”(TimerService)
,我们可以通过它访问流中的事件(event)
、时间戳(timestamp)
、水位线(watermark)
,甚至可以注册“定时事件”
。而且处理函数继承了 AbstractRichFunction
抽象类,所以拥有富函数类的所有特性,同样可以访问状态(state)
和其他运行时信息。此外,处理函数还可以直接将数据输出到侧输出流(side output)
中。所以,处理函数是最为灵活的处理方法,可以实现各种自定义的业务逻辑;同时也是整个 DataStream API 的底层基础。
处理函数的使用与基本的转换操作类似,只需要直接基于 DataStream
调用.process()
方法就可以了。方法需要传入一个 ProcessFunction
作为参数,用来定义处理逻辑。
stream.process(new MyProcessFunction())
这里 ProcessFunction
不是接口,而是一个抽象类,继承了 AbstractRichFunction
;MyProcessFunction
是它的一个具体实现。所以所有的处理函数,都是富函数(RichFunction)
,富函数可以调用的东西这里同样都可以调用。
代码实例:
public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);SingleOutputStreamOperator<Event> stream = env.addSource(new CustomSource()).assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner(new SerializableTimestampAssigner<Event>() {@Overridepublic long extractTimestamp(Event element, long recordTimestamp) {return element.getTimestamp();}}));stream.process(new ProcessFunction<Event, String>() {@Overridepublic void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {out.collect(value.toString());if (value.getUser().equals("Mary")) {out.collect(value.user + "click " + value.getUrl());} else if (value.getUser().equals("Alice")) {out.collect(value.user);out.collect(value.user);}System.out.println("timestamp:" + ctx.timestamp());System.out.println("watermark:" + ctx.timerService().currentWatermark());System.out.println(getRuntimeContext().getIndexOfThisSubtask());}@Overridepublic void onTimer(long timestamp, ProcessFunction<Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {super.onTimer(timestamp, ctx, out);}@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);}@Overridepublic void close() throws Exception {super.close();}}).print();env.execute();}
运行结果:
这里第一次的水位线的值其实是个默认值,Long.MIN_VALUE + outOfOrdernessMillis + 1;
然后每次下一次的水位线都是上一次的timestamp - 1
2.1.2 ProcessFunction 解析
抽象类 ProcessFunction 继承了 AbstractRichFunction,有两个泛型类型参数:I 表示 Input,也就是输入的数据类型;O 表示 Output,也就是处理完成之后输出的数据类型。
内部单独定义了两个方法:一个是必须要实现的抽象方法.processElement();另一个是非抽象方法.onTimer()。
public abstract class ProcessFunction<I, O> extends AbstractRichFunction {...
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 {}
...
}
2.1.2.1 抽象方法.processElement()
用于处理元素,定义了处理的核心逻辑。这个方法对流中的每个元素都会调用一次,参数包括三个: 输入数据值 value
,上下文 ctx
,以及“收集器”(Collector)out
。方法没有返回值,处理之后的输出数据是通过收集器 out 来定义的。
value
: 当前流中的输入元素,也就是正在处理的数据,类型与流中数据类型一致。cts
:类型是ProcessFunction
中定义的内部抽象类Context
,表示当前运行的上下文,可以获取到当前的时间戳,并提供了用于查询时间和注册定时器的“定时服务”(TimerService)
,以及可以将数据发送到“侧输出流”(side output)
的方法.output()
。
Context 抽象类定义如下:
public abstract class Context {public abstract Long timestamp();public abstract TimerService timerService();public abstract <X> void output(OutputTag<X> outputTag, X value);
}
out
:“收集器”(类型为 Collector)
,用于返回输出数据。使用方式与flatMap
算子中的收集器完全一样,直接调用out.collect()
方法就可以向下游发出一个数据。这个方法可以多次调用,也可以不调用。
ProcessFunction
可以轻松实现flatMap
这样的基本转换功能(当然 map
、filter
更不在话下);而通过富函数提供的获取上下文方法.getRuntimeContext()
,也可以自定义状态(state)
进行处理,这也就能实现聚合
操作的功能了。
2.1.2.2 非抽象方法.onTimer()
@Override
public void onTimer(long timestamp, ProcessFunction<Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {super.onTimer(timestamp, ctx, out);
}
用于定义定时触发的操作,这是一个非常强大、也非常有趣的功能。这个方法只有在注册好的定时器触发的时候才会调用,而定时器是通过“定时服务”TimerService
来注册的。打个比方,注册定时器(timer)
就是设了一个闹钟,到了设定时间就会响;而.onTimer()
中定义的,就是闹钟响的时候要做的事
。所以它本质上是一个基于时间的“回调”(callback)
方法,通过时间的进展来触发;在事件时间语义下就是由水位线(watermark)
来触发了。
与.processElement()
类似,定时方法.onTimer()
也有三个参数:时间戳(timestamp)
,上下文(ctx)
,以及收集器(out)
。这里的timestamp
是指设定好的触发时间,事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器,所以也可以调用定时服务(TimerService)
,以及任意输出处理
之后的数据。
既然有.onTimer()
方法做定时触发,我们用ProcessFunction
也可以自定义数据按照时间分组、定时触发计算输出结果;这其实就实现了窗口(window)
的功能。所以说 ProcessFunction
是真正意义上的终极奥义,用它可以实现一切功能。
处理函数都是基于事件触发
的。水位线就如同插入流中的一条数据一样;只不过处理真正的数据事件调用的是.processElement()
方法,而处理水位线事件调用的是.onTimer()
。
.onTimer()
方法只是定时器触发时的操作,而定时器(timer)
真正的设置需要用到上下文 ctx
中的定时服务。在Flink
中,只有“按键分区流”KeyedStream
才支持设置定时器
的操作,所以之前的代码中并没有使用定时器。所以基于不同类型的流,可以使用不同的处理函数。