十一、处理函数
之前所介绍的流处理API，无论是基本的转换、聚合，还是更为复杂的窗口操作，其实都是基于DataStream进行转换的；所以可以统称为DataStream API，这也是Flink编程的核心。而我们知道，为了让代码有更强大的表现力和易用性，Flink本身提供了多层API，DataStream API只是中间的一环，如图所示：
在更底层，我们可以不定义任何具体的算子（比如map，filter，或者window），而只是提炼出一个统一的“处理”（process）操作——它是所有转换算子的一个概括性的表达，可以自定义处理逻辑，所以这一层接口就被叫作“处理函数”（process function）。
11.1 基本处理函数（ProcessFunction）
处理函数主要是定义数据流的转换操作，所以也可以把它归到转换算子中。我们知道在Flink中几乎所有转换算子都提供了对应的函数类接口，处理函数也不例外；它所对应的函数类，就叫作ProcessFunction。
11.1.1 处理函数的功能和使用
我们之前学习的转换算子，一般只是针对某种具体操作来定义的，能够拿到的信息比较有限。如果我们想要访问事件的时间戳，或者当前的水位线信息，都是完全做不到的。跟时间相关的操作，目前我们只会用窗口来处理。而在很多应用需求中，要求我们对时间有更精细的控制，需要能够获取水位线，甚至要“把控时间”、定义什么时候做什么事，这就不是基本的时间窗口能够实现的了。
这时就需要使用底层的处理函数（ProcessFunction）。处理函数提供了一个“定时服务”（TimerService），我们可以通过它访问流中的事件（event）、时间戳（timestamp）、水位线（watermark），甚至可以注册“定时事件”。而且处理函数继承了AbstractRichFunction抽象类，所以拥有富函数类的所有特性，同样可以访问状态（state）和其他运行时信息。此外，处理函数还可以直接将数据输出到侧输出流（side output）中。所以，处理函数是最为灵活的处理方法，可以实现各种自定义的业务逻辑；同时也是整个DataStream API的底层基础。处理函数的使用与基本的转换操作类似，只需要直接基于DataStream调用.process()方法就可以了。方法需要传入一个ProcessFunction作为参数，用来定义处理逻辑。
stream.process(new MyProcessFunction()) 这里ProcessFunction不是接口，而是一个抽象类，继承了AbstractRichFunction；MyProcessFunction是它的一个具体实现。所以所有的处理函数，都是富函数（RichFunction），富函数可以调用的东西这里同样都可以调用。
11.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 {} ... }
1. 抽象方法.processElement()
用于“处理元素”，定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次，参数包括三个：输入数据值value，上下文ctx，以及“收集器”（Collector）out。方法没有返回值，处理之后的输出数据是通过收集器out来定义的。
value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类型一致。 ctx：类型是ProcessFunction中定义的内部抽象类Context，表示当前运行的上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服务”(TimerService)，以及可以将数据发送到“侧输出流”（side output）的方法.output()。 out：“收集器”（类型为Collector），用于返回输出数据。使用方式与flatMap算子中的收集器完全一样，直接调用out.collect()方法就可以向下游发出一个数据。这个方法可以多次调用，也可以不调用。 通过几个参数的分析不难发现，ProcessFunction可以轻松实现flatMap这样的基本转换功能（当然map、filter更不在话下）；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了。
2. 非抽象方法.onTimer()
用于定义定时触发的操作，这是一个非常强大、也非常有趣的功能。这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过“定时服务”TimerService来注册的。打个比方，注册定时器（timer）就是设了一个闹钟，到了设定时间就会响；而.onTimer()中定义的，就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”（callback）方法，通过时间的进展来触发；在事件时间语义下就是由水位线（watermark）来触发了。
与.processElement()类似，定时方法.onTimer()也有三个参数：时间戳（timestamp），上下文（ctx），以及收集器（out）。这里的timestamp是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService），以及任意输出处理之后的数据。既然有.onTimer()方法做定时触发，我们用ProcessFunction也可以自定义数据按照时间分组、定时触发计算输出结果；这其实就实现了窗口（window）的功能。所以说ProcessFunction是真正意义上的终极奥义，用它可以实现一切功能。
这里需要注意的是，上面的.onTimer()方法只是定时器触发时的操作，而定时器（timer）真正的设置需要用到上下文ctx中的定时服务。在Flink中，只有“按键分区流”KeyedStream才支持设置定时器的操作，所以之前的代码中我们并没有使用定时器。所以基于不同类型的流，可以使用不同的处理函数，它们之间还是有一些微小的区别的。接下来我们就介绍一下处理函数的分类。
11.1.3 处理函数的分类
Flink中的处理函数其实是一个大家族，ProcessFunction只是其中一员。
我们知道，DataStream在调用一些转换方法之后，有可能生成新的流类型；例如调用.keyBy()之后得到KeyedStream，进而再调用.window()之后得到WindowedStream。对于不同类型的流，其实都可以直接调用.process()方法进行自定义处理，这时传入的参数就都叫作处理函数。当然，它们尽管本质相同，都是可以访问状态和时间信息的底层API，可彼此之间也会有所差异。
Flink提供了8个不同的处理函数：
（1）ProcessFunction 最基本的处理函数，基于DataStream直接调用.process()时作为参数传入。 （2）KeyedProcessFunction 对流按键分区后的处理函数，基于KeyedStream调用.process()时作为参数传入。要想使用定时器，比如基于KeyedStream。 （3）ProcessWindowFunction 开窗之后的处理函数，也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。 （4）ProcessAllWindowFunction 同样是开窗之后的处理函数，基于AllWindowedStream调用.process()时作为参数传入。 （5）CoProcessFunction 合并（connect）两条流之后的处理函数，基于ConnectedStreams调用.process()时作为参数传入。关于流的连接合并操作，我们会在后续章节详细介绍。 （6）ProcessJoinFunction 间隔连接（interval join）两条流之后的处理函数，基于IntervalJoined调用.process()时作为参数传入。 （7）BroadcastProcessFunction 广播连接流处理函数，基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream，是一个未keyBy的普通DataStream与一个广播流（BroadcastStream）做连接（conncet）之后的产物。关于广播流的相关操作，我们会在后续章节详细介绍。 （8）KeyedBroadcastProcessFunction 按键分区的广播连接流处理函数，同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是，这时的广播连接流，是一个KeyedStream与广播流（BroadcastStream）做连接之后的产物。 接下来，我们就对KeyedProcessFunction和ProcessWindowFunction的具体用法展开详细说明。 11.2 按键分区处理函数（KeyedProcessFunction） 在上节中提到，只有在KeyedStream中才支持使用TimerService设置定时器的操作。所以一般情况下，我们都是先做了keyBy分区之后，再去定义处理操作；代码中更加常见的处理函数是KeyedProcessFunction。
11.2.1定时器（Timer）和定时服务（TimerService）
KeyedProcessFunction的一个特色，就是可以使用定时器。
定时器（timers）是处理函数中进行时间相关操作的主要机制。在.onTimer()方法中可以实现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能，是通过上下文中提供的“定时服务”（TimerService）来实现的。定时服务与当前运行的环境有关。前面已经介绍过，ProcessFunction的上下文（Context）中提供了.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会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个key和时间戳，最多只有一个定时器，如果注册了多次，onTimer（）方法也将只被调用一次。这样一来，我们在代码中就方便了很多，可以肆无忌惮地对一个key注册定时器，而不用担心重复定义——因为一个时间戳上的定时器只会触发一次。
11.2.2 KeyedProcessFunction的使用
基于keyBy之后的KeyedStream，直接调用.process()方法，这时需要传入的参数就是KeyedProcessFunction的实现类。
stream.keyBy( t -> t.f0 ) .process(new MyKeyedProcessFunction())
类似地，KeyedProcessFunction也是继承自AbstractRichFunction的一个抽象类，与ProcessFunction的定义几乎完全一样，区别只是在于类型参数多了一个K，这是当前按键分区的key的类型。同样地，我们必须实现一个.processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法.onTimer()，用来定义定时器触发时的回调操作。
下面是一个使用定时器的具体示例：
public class ProcessingTimeTimerExample { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env .addSource(new CustomSource()) .keyBy(r -> true) .process(new KeyedProcessFunction<Boolean, String, String>() { // 每来一条数据都会调用一次 @Override public void processElement(String s, Context context, Collector<String> collector) throws Exception { long currTs = context.timerService().currentProcessingTime(); collector.collect("数据到达，到达时间是：" + new Timestamp(currTs)); // 注册10s之后的定时器 context.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(); } public static class CustomSource implements SourceFunction<String> { @Override public void run(SourceFunction.SourceContext<String> ctx) throws Exception { ctx.collect("a"); // 为了让程序不直接退出，等待20秒 Thread.sleep(20 * 1000L); } @Override public void cancel() { } } }
在上面的代码中，我们单独定义了一个数据源。这个数据源只发出一条数据，之后等待20秒，这样程序就不会在处理完数据之后直接退出，方便我们看到定时器的行为。由于定时器只能在KeyedStream上使用，所以先要进行keyBy；这里的.keyBy(r -> true)是将所有数据的key都指定为了true，其实就是所有数据拥有相同的key，会分配到同一个分区。
之后我们自定义了一个KeyedProcessFunction，其中.processElement()方法是每来一个数据都会调用一次，主要是定义了一个10秒之后的定时器；而.onTimer()方法则会在定时器触发时调用。所以我们会看到，程序运行后先在控制台输出“数据到达”的信息，等待10秒之后，又会输出“定时器触发”的信息，打印出的时间间隔正是10秒。
11.3 窗口处理函数
除了KeyedProcessFunction，另外一大类常用的处理函数，就是基于窗口的ProcessWindowFunction和ProcessAllWindowFunction了。在第六章窗口函数的介绍中，我们之前已经简单地使用过窗口处理函数了。
11.3.1 窗口处理函数的使用
进行窗口计算，我们可以直接调用现成的简单聚合方法（sum/max/min）,也可以通过调用.reduce()或.aggregate()来自定义一般的增量聚合函数（ReduceFunction/AggregateFucntion）；而对于更加复杂、需要窗口信息和额外状态的一些场景，我们还可以直接使用全窗口函数、把数据全部收集保存在窗口内，等到触发窗口计算时再统一处理。窗口处理函数就是一种典型的全窗口函数。
窗口处理函数ProcessWindowFunction的使用与其他窗口函数类似，也是基于WindowedStream直接调用方法就可以，只不过这时调用的是.process()。
stream.keyBy( t -> t.f0 ) .window( TumblingEventTimeWindows.of(Time.seconds(10)) ) .process(new MyProcessWindowFunction())
11.3.2 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()方法中进行显式地清除，避免内存溢出。
至于另一种窗口处理函数ProcessAllWindowFunction，它的用法非常类似。区别在于它基于的是AllWindowedStream，相当于对没有keyBy的数据流直接开窗并调用.process()方法:
stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) ) .process(new MyProcessAllWindowFunction())
11.4 应用案例——Top N
窗口的计算处理，在实际应用中非常常见。对于一些比较复杂的需求，如果增量聚合函数无法满足，我们就需要考虑使用窗口处理函数这样的“大招”了。网站中一个非常经典的例子，就是实时统计一段时间内的热门url。例如，需要统计最近10秒钟内最热门的两个url链接，并且每5秒钟更新一次。我们知道，这可以用一个滑动窗口来实现，而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集url的访问数据，按照不同的url进行统计，而后汇总排序并最终输出前两名。这其实就是著名的“Top N”问题。很显然，简单的增量聚合可以得到url链接的访问量，但是后续的排序输出Top N就很难实现了。所以接下来我们用窗口处理函数进行实现。
11.4.1 使用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> pvStream = env .addSource(new ClickSource()) .assignTimestampsAndWatermarks( WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(0)) .withTimestampAssigner(new SerializableTimestampAssigner<Event>() { @Override public long extractTimestamp(Event element, long recordTimestamp) { return element.timestamp; } }) ); pvStream .map(new MapFunction<Event, String>() { @Override public String map(Event value) throws Exception { return value.url; } }) // 只需要url，提取转换成String .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); result .append("浏览量No." + (i + 1) + " ") .append("url：" + temp.f0 + " ") .append("浏览量：" + temp.f1 + " ") .append("窗口结束时间：" + new Timestamp(context.window().getEnd()) + "\n"); } result .append("========================================\n\n\n"); out.collect(result.toString()); } }) .print(); env.execute(); } }
11.4.2 使用KeyedProcessFunction
在上一小节的实现过程中，我们没有进行按键分区，直接将所有数据放在一个分区上进行了开窗操作。这相当于将并行度强行设置为1，在实际应用中是要尽量避免的，所以Flink官方也并不推荐使用AllWindowedStream进行处理。另外，我们在全窗口函数中定义了HashMap来统计url链接的浏览量，计算过程是要先收集齐所有数据、然后再逐一遍历更新HashMap，这显然不够高效。基于这样的想法，我们可以从两个方面去做优化：一是对数据进行按键分区，分别统计浏览量；二是进行增量聚合，得到结果最后再做排序输出。所以，我们可以使用增量聚合函数AggregateFunction进行浏览量的统计，然后结合ProcessWindowFunction排序输出来实现Top N的需求。具体实现可以分成两步：先对每个url链接统计出浏览量，然后再将统计结果收集起来，排序输出最终结果。由于最后的排序还是基于每个时间窗口的，所以为了让输出的统计结果中包含窗口信息，我们可以借用第六章中定义的POJO类UrlViewCount来表示，它包含了url、浏览量（count）以及窗口的起始结束时间。之后对UrlViewCount的处理，可以先按窗口分区，然后用KeyedProcessFunction来实现。
总结处理流程如下： （1）读取数据源； （2）筛选浏览行为（pv）； （3）提取时间戳并生成水位线； （4）按照url进行keyBy分区操作； （5）开长度为1小时、步长为5分钟的事件时间滑动窗口； （6）使用增量聚合函数AggregateFunction，并结合全窗口函数WindowFunction进行窗口聚合，得到每个url、在每个统计窗口内的浏览量，包装成UrlViewCount； （7）按照窗口进行keyBy分区操作； （8）对同一窗口的统计结果数据，使用KeyedProcessFunction进行收集并排序输出。 用KeyedProcessFunction来收集数据做排序，这时面对的是窗口聚合之后的数据流，而窗口已经不存在了；我们需要确保能够收集齐所有数据，所以应该在窗口结束时间基础上再“多等一会儿”。具体实现上，可以采用一个延迟触发的事件时间定时器。基于窗口的结束时间来设定延迟，其实并不需要等太久——因为我们是靠水位线的推进来触发定时器，而水位线的含义就是“之前的数据都到齐了”。所以我们只需要设置1毫秒的延迟，就一定可以保证这一点。而在等待过程中，之前已经到达的数据应该缓存起来，我们这里用一个自定义的“列表状态”（ListState）来进行存储，如图所示。这个状态需要使用富函数类的getRuntimeContext()方法获取运行时上下文来定义，我们一般把它放在open()生命周期方法中。之后每来一个UrlViewCount，就把它添加到当前的列表状态中，并注册一个触发时间为窗口结束时间加1毫秒（windowEnd + 1）的定时器。待到水位线到达这个时间，定时器触发，我们可以保证当前窗口所有url的统计结果UrlViewCount都到齐了；于是从状态中取出进行排序输出。
具体代码实现如下：
public class KeyedProcessTopN { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); SingleOutputStreamOperator<Event> pvStream = env .addSource(new ClickSource()) .assignTimestampsAndWatermarks( WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(0)) .withTimestampAssigner(new SerializableTimestampAssigner<Event>() { @Override public long extractTimestamp(Event element, long recordTimestamp) { return element.timestamp; } }) ); // 先求出每个url在每个窗口的浏览量 SingleOutputStreamOperator<UrlViewCount> uvcStream = pvStream .keyBy(r -> r.url) .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))) .aggregate(new CountAgg(), new WindowResult()); // 增量聚合，并结合全窗口函数包装UrlViewCount // 针对同一个窗口中的不同url的UrlViewCount次数，进行排序输出 KeyedStream<UrlViewCount, Long> uvcKeyedStream = uvcStream .keyBy(r -> r.windowEnd); uvcKeyedStream .process(new TopN(2)) .print(); env.execute(); } public static class TopN extends KeyedProcessFunction<Long, UrlViewCount, String> { // 列表状态变量 private ListState<UrlViewCount> UrlViewCountListState; private Integer threshold; public TopN(Integer threshold) { this.threshold = threshold; } @Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 声明一个列表状态，保存已经到达的统计结果 UrlViewCountListState = getRuntimeContext().getListState( new ListStateDescriptor<UrlViewCount>("list-state", Types.POJO(UrlViewCount.class)) ); } @Override public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception { // 添加到列表状态变量中 UrlViewCountListState.add(value); // 水位线达到 窗口结束时间 + 1毫秒 时触发定时器来进行排序 ctx.timerService().registerEventTimeTimer(value.windowEnd + 1); } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception { super.onTimer(timestamp, ctx, out); // 将数据从列表状态变量中取出，放入ArrayList ArrayList<UrlViewCount> UrlViewCountArrayList = new ArrayList<>(); for (UrlViewCount uvc : UrlViewCountListState.get()) { UrlViewCountArrayList.add(uvc); } // 清空状态释放资源 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"); for (int i = 0; i < this.threshold; i++) { UrlViewCount UrlViewCount = UrlViewCountArrayList.get(i); result .append("浏览量No." + (i + 1) + " ") .append("url：" + UrlViewCount.url + " ") .append("浏览量：" + UrlViewCount.count + " ") .append("窗口结束时间：" + new Timestamp(timestamp - 1) + "\n"); } result .append("========================================\n\n\n"); out.collect(result.toString()); } } public static class CountAgg implements AggregateFunction<Event, Long, Long> { @Override public Long createAccumulator() { return 0L; } @Override public Long add(Event value, Long accumulator) { return accumulator + 1L; } @Override public Long getResult(Long accumulator) { return accumulator; } @Override public Long merge(Long a, Long b) { return null; } } public static class WindowResult extends ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow> { @Override public void process(String s, Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception { out.collect(new UrlViewCount(s, elements.iterator().next(), context.window().getStart(), context.window().getEnd())); } } public static class UrlViewCount { public String url; public Long count; public Long windowStart; public Long windowEnd; public UrlViewCount() { } public UrlViewCount(String url, Long count, Long windowStart, Long windowEnd) { this.url = url; this.count = count; this.windowStart = windowStart; this.windowEnd = windowEnd; } @Override public String toString() { return "UrlViewCount{" + "url='" + url + '\'' + ", count=" + count + ", windowStart=" + new Timestamp(windowStart) + ", windowEnd=" + new Timestamp(windowEnd) + '}'; } } }
我们在上面的代码中使用了后面要讲解的ListState。这里可以先简单说明一下。我们先声明一个列表状态变量:
private ListState<Event> UrlViewCountListState;
然后在open方法中初始化了列表状态变量，我们初始化的时候使用了ListStateDescriptor描述符，这个描述符用来告诉Flink列表状态变量的名字和类型。列表状态变量是单例，也就是说只会被实例化一次。这个列表状态变量的作用域是当前key所对应的逻辑分区。我们使用add方法向列表状态变量中添加数据，使用get方法读取列表状态变量中的所有元素。
11.5 侧输出流（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);
