文章目录
- 处理函数
- 基本处理函数(ProcessFunction)
- 功能和使用
- ProcessFunction 解析
- 分类
- 按键分区处理函数(KeyedProcessFunction)
- 定时器(Timer)和定时服务(TimerService)
- KeyedProcessFunction 的使用
- 窗口处理函数
- 窗口处理函数的使用
- ProcessWindowFunction 解析
- 应用案例——Top N
- 使用 ProcessAllWindowFunction
- 使用 KeyedProcessFunction
- 侧输出流(Side Output)
处理函数
之前所介绍的流处理 API,无论是基本的转换、聚合,还是更为复杂的窗口操作,其实都是基于 DataStream 进行转换的;所以可以统称为 DataStream API,这也是 Flink 编程的核心。而我们知道,为了让代码有更强大的表现力和易用性,Flink 本身提供了多层 API,DataStream API 只是中间的一环。
在更底层,我们可以不定义任何具体的算子(比如 map,filter,或者 window),而只是提炼出一个统一的“处理”(process)操作——它是所有转换算子的一个概括性的表达,可以自定义处理逻辑,所以这一层接口就被叫作“处理函数”(process function)。
在处理函数中,我们直面的就是数据流中最基本的元素:数据事件(event)、状态(state)以及时间(time)。这就相当于对流有了完全的控制权。处理函数比较抽象,没有具体的操作,所以对于一些常见的简单应用(比如求和、开窗口)会显得有些麻烦;不过正是因为它不限定具体做什么,所以理论上我们可以做任何事情,实现所有需求。
基本处理函数(ProcessFunction)
功能和使用
我们之前学习的转换算子,一般只是针对某种具体操作来定义的,能够拿到的信息比较有限。比如 map 算子,我们实现的 MapFunction 中,只能获取到当前的数据,定义它转换之后的形式;而像窗口聚合这样的复杂操作,AggregateFunction 中除数据外,还可以获取到当前的状态(以累加器 Accumulator 形式出现)。另外我们还介绍过富函数类,比如 RichMapFunction,它提供了获取运行时上下文的方法 getRuntimeContext(),可以拿到状态,还有并行度、任务名称之类的运行时信息。
但是无论那种算子,如果我们想要访问事件的时间戳,或者当前的水位线信息,都是完全做不到的。在定义生成规则之后,水位线会源源不断地产生,像数据一样在任务间流动,可我们却不能像数据一样去处理它;跟时间相关的操作,目前我们只会用窗口来处理。而在很多应用需求中,要求我们对时间有更精细的控制,需要能够获取水位线,甚至要“把控时间”、定义什么时候做什么事,这就不是基本的时间窗口能够实现的了。
于是必须祭出大招——处理函数(ProcessFunction)了。处理函数提供了一个“定时服务”(TimerService),我们可以通过它访问流中的事件(event)、时间戳(timestamp)、水位线(watermark),甚至可以注册“定时事件”。而且处理函数继承了 AbstractRichFunction 抽象类,所以拥有富函数类的所有特性,同样可以访问状态(state)和其他运行时信息。此外,处理函数还可以直接将数据输出到侧输出流(side output)中。所以,处理函数是最为灵活的处理方法,可以实现各种自定义的业务逻辑;同时也是整个 DataStream API 的底层基础。
处理函数的使用与基本的转换操作类似,只需要直接基于 DataStream 调用.process()方法就可以了。方法需要传入一个 ProcessFunction 作为参数,用来定义处理逻辑。
stream.process(new MyProcessFunction())
这里 ProcessFunction 不是接口,而是一个抽象类,继承了 AbstractRichFunction;MyProcessFunction 是它的一个具体实现。所以所有的处理函数,都是富函数(RichFunction),富函数可以调用的东西这里同样都可以调用。
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()方法,自定义了一种处理逻辑:当数据的 user 为“Mary”时,将其输出一次;而如果为“Bob”时,将 user 输出两次。这里的输出 , 是通过调用 out.collect() 来实现的。另外我们还可以调用ctx.timerService().currentWatermark() 来 获 取 当 前 的 水 位 线 打 印 输 出 。 所 以 可 以 看 到 ,ProcessFunction 函数有点像 FlatMapFunction 的升级版。可以实现 Map、Filter、FlatMap 的所有功能。很明显,处理函数非常强大,能够做很多之前做不到的事情。
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 {}
...
}
(1)抽象方法.processElement()
用于“处理元素”,定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次,参数包括三个:输入数据值 value,上下文 ctx,以及“收集器”(Collector)out。方法没有返回值,处理之后的输出数据是通过收集器 out 来定义的。
-
value:当前流中的输入元素,也就是正在处理的数据,类型与流中数据类型一致。
-
ctx:类型是 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()方法就可以向下游发出一个数据。这个方法可以多次调用,也可以不调用。
(2)非抽象方法.onTimer()
用于定义定时触发的操作,这是一个非常强大、也非常有趣的功能。这个方法只有在注册好的定时器触发的时候才会调用,而定时器是通过“定时服务”TimerService 来注册的。打个比方,注册定时器(timer)就是设了一个闹钟,到了设定时间就会响;而.onTimer()中定义的,就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”(callback)方法,通过时间的进展来触发;在事件时间语义下就是由水位线(watermark)来触发了。
与.processElement()类似,定时方法.onTimer()也有三个参数:时间戳(timestamp),上下文(ctx),以及收集器(out)。这里的 timestamp 是指设定好的触发时间(即通过ctx.timeService().register…设置的时间,事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器,所以也可以调用定时服务(TimerService),以及任意输出处理之后的数据。
分类
Flink 中的处理函数其实是一个大家族,ProcessFunction 只是其中一员。
我们知道,DataStream 在调用一些转换方法之后,有可能生成新的流类型;例如调用.keyBy()之后得到 KeyedStream,进而再调用.window()之后得到 WindowedStream。对于不同类型的流,其实都可以直接调用.process()方法进行自定义处理,这时传入的参数就都叫作处理函数。
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)做连接之后的产物。
按键分区处理函数(KeyedProcessFunction)
定时器(Timer)和定时服务(TimerService)
KeyedProcessFunction 的一个特色,就是可以灵活地使用定时器。
定时器(timers)是处理函数中进行时间相关操作的主要机制。在.onTimer()方法中可以实现定时处理的逻辑,而它能触发的前提,就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能,是通过上下文中提供的“定时服务”(TimerService)来实现的。
ProcessFunction 的上下文(Context)中提供了.timerService()方法,可以直接返回一个 TimerService 对象:
public abstract TimerService timerService();
TimerService 是 Flink 关于时间和定时器的基础服务接口,包含以下六个方法:
// 获取当前的处理时间
long currentProcessingTime();
// 获取当前的水位线(事件时间)
long currentWatermark();
// 注册处理时间定时器,当处理时间超过 time 时触发
void registerProcessingTimeTimer(long time);
// 注册事件时间定时器,当水位线超过 time 时触发
void registerEventTimeTimer(long time);
// 删除触发时间为 time 的处理时间定时器
void deleteProcessingTimeTimer(long time);
// 删除触发时间为 time 的处理时间定时器
void deleteEventTimeTimer(long time);
六个方法可以分成两大类:基于处理时间和基于事件时间。而对应的操作主要有三个:获取当前时间,注册定时器,以及删除定时器。需要注意,尽管处理函数中都可以直接访问TimerService,不过只有基于 KeyedStream 的处理函数,才能去调用注册和删除定时器的方法;未作按键分区的 DataStream 不支持定时器操作,只能获取当前时间。
对于处理时间和事件时间这两种类型的定时器,TimerService 内部会用一个优先队列将它们的时间戳(timestamp)保存起来,排队等待执行。可以认为,定时器其实是 KeyedStream上处理算子的一个状态,它以时间戳作为区分。所以 TimerService 会以键(key)和时间戳为标准,对定时器进行去重;也就是说对于每个 key 和时间戳,最多只有一个定时器,如果注册了多次,onTimer()方法也将只被调用一次。
Flink 对.onTimer()和.processElement()方法是同步调用的(synchronous),所以不会出现状态的并发修改。
Flink 的定时器同样具有容错性,它和状态一起都会被保存到一致性检查点(checkpoint)中。当发生故障时,Flink 会重启并读取检查点中的状态,恢复定时器。如果是处理时间的定时器,有可能会出现已经“过期”的情况,这时它们会在重启时被立刻触发。
KeyedProcessFunction 的使用
KeyedProcessFunction 可以说是处理函数中的“嫡系部队”,可以认为是 ProcessFunction 的一个扩展。我们只要基于 keyBy 之后的 KeyedStream,直接调用.process()方法,这时需要传入的参数就是 KeyedProcessFunction 的实现类。
stream.keyBy( t -> t.f0 ).process(new MyKeyedProcessFunction())
类似地,KeyedProcessFunction 也是继承自 AbstractRichFunction 的一个抽象类,源码中定义如下:
public abstract class KeyedProcessFunction<K, 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 {}
public abstract class Context {...}
...
}
可以看到与 ProcessFunction 的定义几乎完全一样,区别只是在于类型参数多了一个 K,这是当前按键分区的 key 的类型。
public class ProcessingTimeTimerTest {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 处理时间语义,不需要分配时间戳和watermark
SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource());
// 要用定时器,必须基于KeyedStream
stream.keyBy(data -> true)
.process(new KeyedProcessFunction<Boolean, Event, String>() {
@Override
public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
Long currTs = ctx.timerService().currentProcessingTime();
out.collect("数据到达,到达时间:" + new Timestamp(currTs));
// 注册一个10秒后的定时器
ctx.timerService().registerProcessingTimeTimer(currTs + 10 * 1000L);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
out.collect("定时器触发,触发时间:" + new Timestamp(timestamp));
}
})
.print();
env.execute();
}
}
在上面的代码中,由于定时器只能在 KeyedStream 上使用,所以先要进行 keyBy;这里的.keyBy(data -> true)是将所有数据的 key 都指定为了 true,其实就是所有数据拥有相同的 key,会分配到同一个分区。
之后我们自定义了一个 KeyedProcessFunction,其中.processElement()方法是每来一个数据都会调用一次,主要是定义了一个 10 秒之后的定时器;而.onTimer()方法则会在定时器触发时调用。所以我们会看到,程序运行后先在控制台输出“数据到达”的信息,等待 10 秒之后,又会输出“定时器触发”的信息,打印出的时间间隔正是 10 秒。
当然,上面的例子是处理时间的定时器,所以我们是真的需要等待 10 秒才会看到结果。事件时间语义下,又会有什么不同呢?
public class EventTimeTimerTest {
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>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.timestamp;
}
}));
// 基于KeyedStream定义事件时间定时器
stream.keyBy(data -> true)
.process(new KeyedProcessFunction<Boolean, Event, String>() {
@Override
public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
out.collect("数据到达,时间戳为:" + ctx.timestamp());
out.collect("数据到达,水位线为:" + ctx.timerService().currentWatermark() + "\n -------分割线-------");
// 注册一个10秒后的定时器
ctx.timerService().registerEventTimeTimer(ctx.timestamp() + 10 * 1000L);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
out.collect("定时器触发,触发时间:" + timestamp);
}
})
.print();
env.execute();
}
// 自定义测试数据源
public static class CustomSource implements SourceFunction<Event> {
@Override
public void run(SourceContext<Event> ctx) throws Exception {
// 直接发出测试数据
ctx.collect(new Event("Mary", "./home", 1000L));
// 为了更加明显,中间停顿5秒钟
Thread.sleep(5000L);
// 发出10秒后的数据
ctx.collect(new Event("Mary", "./home", 11000L));
Thread.sleep(5000L);
// 发出10秒+1ms后的数据
ctx.collect(new Event("Alice", "./cart", 11001L));
Thread.sleep(5000L);
}
@Override
public void cancel() { }
}
}
由于是事件时间语义,所以我们必须从数据中提取出数据产生的时间戳。这里为了更清楚地看到程序行为,我们自定义了一个数据源,发出三条测试数据,时间戳分别为 1000、11000和 11001,并且发出数据后都会停顿 5 秒。
在代码中,我们依然将所有数据分到同一分区,然后在自定义的 KeyedProcessFunction 中使用定时器。同样地,每来一条数据,我们就将当前的数据时间戳和水位线信息输出,并注册一个 10 秒后(以当前数据时间戳为基准)的事件时间定时器。执行程序结果如下:
数据到达,时间戳为:1000
数据到达,水位线为:-9223372036854775808
-------分割线-------
数据到达,时间戳为:11000
数据到达,水位线为:999
-------分割线-------
数据到达,时间戳为:11001
数据到达,水位线为:10999
-------分割线-------
定时器触发,触发时间:11000
定时器触发,触发时间:21000
定时器触发,触发时间:21001
每来一条数据,都会输出两行“数据到达”的信息,并以分割线隔开;两条数据到达的时间间隔为 5 秒。当第三条数据到达后,随后立即输出一条定时器触发的信息;再过 5 秒之后,剩余两条定时器信息输出,程序运行结束。
我们可以发现,数据到来之后,当前的水位线与时间戳并不是一致的。当第一条数据到来,时间戳为 1000,可水位线的生成是周期性的(默认 200ms 一次),不会立即发生改变,所以依然是最小值 Long.MIN_VALUE;随后只要到了水位线生成的时间点(200ms 到了),就会依据当前的最大时间戳 1000 来生成水位线了。这里我们没有设置水位线延迟,默认需要减去 1 毫秒,所以水位线推进到了 999。而当时间戳为 11000 的第二条数据到来之后,水位线同样没有立即改变,仍然是 999,就好像总是“滞后”数据一样。
窗口处理函数
除 了 KeyedProcessFunction , 另 外 一 大 类 常 用 的 处 理 函 数 , 就 是 基 于 窗 口 的ProcessWindowFunction 和 ProcessAllWindowFunction 了。
窗口处理函数的使用
进行窗口计算,我们可以直接调用现成的简单聚合方法(sum/max/min),也可以通过调用.reduce()或.aggregate()来自定义一般的增量聚合函数(ReduceFunction/AggregateFucntion);而对于更加复杂、需要窗口信息和额外状态的一些场景,我们还可以直接使用全窗口函数、把数据全部收集保存在窗口内,等到触发窗口计算时再统一处理。窗口处理函数就是一种典型的全窗口函数。
窗 口 处 理 函 数 ProcessWindowFunction 的使用与其他窗口函数类似 ,也是基于WindowedStream 直接调用方法就可以,只不过这时调用的是.process()。
stream.keyBy( t -> t.f0 )
.window( TumblingEventTimeWindows.of(Time.seconds(10)) )
.process(new MyProcessWindowFunction())
ProcessWindowFunction 解析
ProcessWindowFunction 既是处理函数又是全窗口函数。从名字上也可以推测出,它的本质似乎更倾向于“窗口函数”一些。事实上它的用法也确实跟其他处理函数有很大不同。我们可以从源码中的定义看到这一点:
public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window>
extends AbstractRichFunction {
...
public abstract void process(KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws Exception;
public void clear(Context context) throws Exception {}
public abstract class Context implements java.io.Serializable {...}
}
ProcessWindowFunction 依然是一个继承了 AbstractRichFunction 的抽象类,它有四个类型参数:
- IN:input,数据流中窗口任务的输入数据类型。
- OUT:output,窗口任务进行计算之后的输出数据类型。
- KEY:数据中键 key 的类型。
- W:窗口的类型,是 Window 的子类型。一般情况下我们定义时间窗口,W就是 TimeWindow。
而内部定义的方法,跟我们之前熟悉的处理函数就有所区别了。因为全窗口函数不是逐个处理元素的,所以处理数据的方法在这里并不是.processElement(),而是改成了.process()。方法包含四个参数。
- key:窗口做统计计算基于的键,也就是之前 keyBy 用来分区的字段。
- context:当前窗口进行计算的上下文,它的类型就是 ProcessWindowFunction内部定义的抽象类 Context。
- elements:窗口收集到用来计算的所有数据,这是一个可迭代的集合类型。
- out:用来发送数据输出计算结果的收集器,类型为 Collector。
可以明显看出,这里的参数不再是一个输入数据,而是窗口中所有数据的集合。而上下文context 所包含的内容也跟其他处理函数有所差别:
public abstract class Context implements java.io.Serializable {
public abstract W window();
public abstract long currentProcessingTime();
public abstract long currentWatermark();
public abstract KeyedStateStore windowState();
public abstract KeyedStateStore globalState();
public abstract <X> void output(OutputTag<X> outputTag, X value);
}
除了可以通过.output()方法定义侧输出流不变外,其他部分都有所变化。
-
这里不再持有TimerService 对象,只能通过 currentProcessingTime()和 currentWatermark()来获取当前时间,所以失去了设置定时器的功能
-
由于当前不是只处理一个数据,所以也不再提供.timestamp()方法。
-
可以通过.window()直接获取到当前的窗口对象,也可以通过.windowState()和.globalState()获取到当前自定义的窗口状态和全局状态。
这里的“窗口状态”是自定义的,不包括窗口本身已经有的状态,针对当前 key、当前窗口有效;而“全局状态”同样是自定义的状态,针对当前 key 的所有窗口有效。
ProcessWindowFunction 中除了.process()方法外,并没有.onTimer()方法,而是多出了一个.clear()方法。从名字就可以看出,这主要是方便我们进行窗口的清理工作。如果我们自定义了窗口状态,那么必须在.clear()方法中进行显式地清除,避免内存溢出。
在触发器中也有一个TriggerContext,它可以起到类似 TimerService 的作用:获取当前时间、注册和删除定时器,另外还可以获取当前的状态。这样设计无疑会让处理流程更加清晰——定时操作也是一种“触发”,所以我们就让所有的触发操作归触发器管,而所有处理数据的操作则归窗口函数管。
至于另一种窗口处理函数 ProcessAllWindowFunction,它的用法非常类似。区别在于它基于的是 AllWindowedStream,相当于对没有 keyBy 的数据流直接开窗并调用.process()方法:
stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) ).process(new MyProcessAllWindowFunction())
应用案例——Top N
网站中一个非常经典的例子,就是实时统计一段时间内的热门 url。例如,需要统计最近10 秒钟内最热门的两个 url 链接,并且每 5 秒钟更新一次。我们知道,这可以用一个滑动窗口
来实现,而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集 url 的访问数据,按照不同的 url 进行统计,而后汇总排序并最终输出前两名。这其实就是著名的“Top N”问题。
很显然,简单的增量聚合可以得到 url 链接的访问量,但是后续的排序输出 Top N 就很难实现了。所以接下来我们用窗口处理函数进行实现。
使用 ProcessAllWindowFunction
一种最简单的想法是,我们干脆不区分 url 链接,而是将所有访问数据都收集起来,统一进行统计计算。所以可以不做 keyBy,直接基于 DataStream 开窗,然后使用全窗口函数ProcessAllWindowFunction 来进行处理。
在窗口中可以用一个 HashMap 来保存每个 url 的访问次数,只要遍历窗口中的所有数据,自然就能得到所有 url 的热门度。最后把 HashMap 转成一个列表 ArrayList,然后进行排序、取出前两名输出就可以了。
public class ProcessAllWindowTopN {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
.assignTimestampsAndWatermarks(
WatermarkStrategy.<Event>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.timestamp;
}
})
);
// 只需要url就可以统计数量,所以转换成String直接开窗统计
SingleOutputStreamOperator<String> result = eventStream
.map(new MapFunction<Event, String>() {
@Override
public String map(Event value) throws Exception {
return value.url;
}
})
.windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))) // 开滑动窗口
.process(new ProcessAllWindowFunction<String, String, TimeWindow>() {
@Override
public void process(Context context, Iterable<String> elements, Collector<String> out) throws Exception {
HashMap<String, Long> urlCountMap = new HashMap<>();
// 遍历窗口中数据,将浏览量保存到一个 HashMap 中
for (String url : elements) {
if (urlCountMap.containsKey(url)) {
long count = urlCountMap.get(url);
urlCountMap.put(url, count + 1L);
} else {
urlCountMap.put(url, 1L);
}
}
ArrayList<Tuple2<String, Long>> mapList = new ArrayList<Tuple2<String, Long>>();
// 将浏览量数据放入ArrayList,进行排序
for (String key : urlCountMap.keySet()) {
mapList.add(Tuple2.of(key, urlCountMap.get(key)));
}
mapList.sort(new Comparator<Tuple2<String, Long>>() {
@Override
public int compare(Tuple2<String, Long> o1, Tuple2<String, Long> o2) {
return o2.f1.intValue() - o1.f1.intValue();
}
});
// 取排序后的前两名,构建输出结果
StringBuilder result = new StringBuilder();
result.append("========================================\n");
for (int i = 0; i < 2; i++) {
Tuple2<String, Long> temp = mapList.get(i);
String info = "浏览量No." + (i + 1) +
" url:" + temp.f0 +
" 浏览量:" + temp.f1 +
" 窗口结束时间:" + new Timestamp(context.window().getEnd()) + "\n";
result.append(info);
}
result.append("========================================\n");
out.collect(result.toString());
}
});
result.print();
env.execute();
}
}
运行结果如下所示:
========================================
浏览量 No.1 url:./prod?id=1 浏览量:2 窗口结束时间:2021-07-01 15:24:25.0
浏览量 No.2 url:./cart 浏览量:1 窗口结束时间:2021-07-01 15:24:25.0
========================================
使用 KeyedProcessFunction
在上面,我们没有进行按键分区,直接将所有数据放在一个分区上进行了开窗操作。这相当于将并行度强行设置为 1,在实际应用中是要尽量避免的,所以 Flink 官方也并不推荐使用 AllWindowedStream 进行处理。另外,我们在全窗口函数中定义了 HashMap来统计 url 链接的浏览量,计算过程是要先收集齐所有数据、然后再逐一遍历更新 HashMap,这显然不够高效。如果我们可以利用增量聚合函数的特性,每来一条数据就更新一次对应 url的浏览量,那么到窗口触发计算时只需要做排序输出就可以了。
基于这样的想法,我们可以从两个方面去做优化:一是对数据进行按键分区,分别统计浏览量;二是进行增量聚合,得到结果最后再做排序输出。所以,我们可以使用增量聚合函数AggregateFunction 进行浏览量的统计,然后结合 ProcessWindowFunction 排序输出来实现 Top N的需求。
具体实现思路就是,先按照 url 对数据进行 keyBy 分区,然后开窗进行增量聚合。这里就会发现一个问题:我们进行按键分区之后,窗口的计算就会只针对当前 key 有效了;也就是说,每个窗口的统计结果中,只会有一个 url 的浏览量,这是无法直接用 ProcessWindowFunction进行排序的。所以我们只能分成两步:
- 先对每个 url 链接统计出浏览量
- 然后再将统计结果收集起来,排序输出最终结果。
因为最后的排序还是基于每个时间窗口的,所以为了让输出的统计结果中包含窗口信息,我们可以借用自定义 POJO 类 UrlViewCount 来表示,它包含了 url、浏览量(count)以及窗口的起始结束时间。之后对 UrlViewCount 的处理,可以先按窗口分区,然后用 KeyedProcessFunction 来实现。
总结处理流程如下:
- 读取数据源;
- 筛选浏览行为(pv);
- 提取时间戳并生成水位线;
- 按照 url 进行 keyBy 分区操作;
- 开长度为 1 小时、步长为 5 分钟的事件时间滑动窗口;
- 使用增量聚合函数 AggregateFunction,并结合全窗口函数 WindowFunction 进行窗口聚合,得到每个 url、在每个统计窗口内的浏览量,包装成 UrlViewCount;
- 按照窗口进行 keyBy 分区操作;
- 对同一窗口的统计结果数据,使用 KeyedProcessFunction 进行收集并排序输出。
public class KeyedProcessTopN {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 从自定义数据源读取数据
SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
.withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
@Override
public long extractTimestamp(Event element, long recordTimestamp) {
return element.timestamp;
}
}));
// 需要按照url分组,求出每个url的访问量
SingleOutputStreamOperator<UrlViewCount> urlCountStream =
eventStream.keyBy(data -> data.url)
.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
.aggregate(new UrlViewCountAgg(),
new UrlViewCountResult());
// 对结果中同一个窗口的统计数据,进行排序处理
SingleOutputStreamOperator<String> result = urlCountStream.keyBy(data -> data.windowEnd)
.process(new TopN(2));
result.print("result");
env.execute();
}
// 自定义增量聚合
public static class UrlViewCountAgg implements AggregateFunction<Event, Long, Long> {
@Override
public Long createAccumulator() {
return 0L;
}
@Override
public Long add(Event value, Long accumulator) {
return accumulator + 1;
}
@Override
public Long getResult(Long accumulator) {
return accumulator;
}
@Override
public Long merge(Long a, Long b) {
return null;
}
}
// 自定义全窗口函数,只需要包装窗口信息
public static class UrlViewCountResult extends ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow> {
@Override
public void process(String url, Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception {
// 结合窗口信息,包装输出内容
Long start = context.window().getStart();
Long end = context.window().getEnd();
out.collect(new UrlViewCount(url, elements.iterator().next(), start, end));
}
}
// 自定义处理函数,排序取top n
public static class TopN extends KeyedProcessFunction<Long, UrlViewCount, String>{
// 将n作为属性
private Integer n;
// 定义一个列表状态
private ListState<UrlViewCount> urlViewCountListState;
public TopN(Integer n) {
this.n = n;
}
@Override
public void open(Configuration parameters) throws Exception {
// 从环境中获取列表状态句柄
urlViewCountListState = getRuntimeContext().getListState(
new ListStateDescriptor<UrlViewCount>("url-view-count-list",
Types.POJO(UrlViewCount.class)));
}
@Override
public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception {
// 将count数据添加到列表状态中,保存起来
urlViewCountListState.add(value);
// 注册 window end + 1ms后的定时器,等待所有数据到齐开始排序
ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
// 将数据从列表状态变量中取出,放入ArrayList,方便排序
ArrayList<UrlViewCount> urlViewCountArrayList = new ArrayList<>();
for (UrlViewCount urlViewCount : urlViewCountListState.get()) {
urlViewCountArrayList.add(urlViewCount);
}
// 清空状态,释放资源
urlViewCountListState.clear();
// 排序
urlViewCountArrayList.sort(new Comparator<UrlViewCount>() {
@Override
public int compare(UrlViewCount o1, UrlViewCount o2) {
return o2.count.intValue() - o1.count.intValue();
}
});
// 取前两名,构建输出结果
StringBuilder result = new StringBuilder();
result.append("========================================\n");
result.append("窗口结束时间:" + new Timestamp(timestamp - 1) + "\n");
for (int i = 0; i < this.n; i++) {
UrlViewCount UrlViewCount = urlViewCountArrayList.get(i);
String info = "No." + (i + 1) + " "
+ "url:" + UrlViewCount.url + " "
+ "浏览量:" + UrlViewCount.count + "\n";
result.append(info);
}
result.append("========================================\n");
out.collect(result.toString());
}
}
}
代码中,我们还利用了定时器的特性:针对同一 key、同一时间戳会进行去重。所以对于同一个窗口而言,我们接到统计结果数据后设定的 windowEnd + 1 的定时器都是一样的,最终只会触发一次计算。而对于不同的 key(这里 key 是 windowEnd),定时器和状态都是独立的,所以我们也不用担心不同窗口间数据的干扰。
我们在上面的代码中使用了后面要讲解的 ListState。这里可以先简单说明一下。我们先声明一个列表状态变量:
private ListState<Event> UrlViewCountListState;
然后在 open 方法中初始化了列表状态变量,我们初始化的时候使用了 ListStateDescriptor描述符,这个描述符用来告诉 Flink 列表状态变量的名字和类型。列表状态变量是单例,也就是说只会被实例化一次。这个列表状态变量的作用域是当前 key 所对应的逻辑分区。我们使用add 方法向列表状态变量中添加数据,使用 get 方法读取列表状态变量中的所有元素。
侧输出流(Side Output)
处理函数还有另外一个特有功能,就是将自定义的数据放入“侧输出流”(side output)输出。这个概念我们并不陌生,之前在讲到窗口处理迟到数据时,最后一招就是输出到侧输出流。而这种处理方式的本质,其实就是处理函数的侧输出流功能。
我们之前讲到的绝大多数转换算子,输出的都是单一流,流里的数据类型只能有一种。而侧输出流可以认为是“主流”上分叉出的“支流”,所以可以由一条流产生出多条流,而且这些流中的数据类型还可以不一样。利用这个功能可以很容易地实现“分流”操作。
具体应用时,只要在处理函数的.processElement()或者.onTimer()方法中,调用上下文的.output()方法就可以了。
DataStream<Integer> stream = env.addSource(...);
SingleOutputStreamOperator<Long> longStream = stream.process(new ProcessFunction<Integer, Long>() {
@Override
public void processElement( Integer value, Context ctx, Collector<Integer>
out) throws Exception {
// 转换成 Long,输出到主流中
out.collect(Long.valueOf(value));
// 转换成 String,输出到侧输出流中
ctx.output(outputTag, "side-output: " + String.valueOf(value));
}
});
这里 output()方法需要传入两个参数,第一个是一个“输出标签”OutputTag,用来标识侧输出流,一般会在外部统一声明;第二个就是要输出的数据。
我们可以在外部先将 OutputTag 声明出来:
OutputTag<String> outputTag = new OutputTag<String>("side-output") {};
如果想要获取这个侧输出流,可以基于处理之后的 DataStream 直接调用.getSideOutput()方法,传入对应的 OutputTag,这个方式与窗口 API 中获取侧输出流是完全一样的。
DataStream<String> stringStream = longStream.getSideOutput(outputTag);