Flink应用案例——Top N
窗口的计算处理，在实际应用中非常常见。对于一些比较复杂的需求，如果增量聚合函数无法满足，我们就需要考虑使用窗口处理函数这样的“大招”了。
网站中一个非常经典的例子，就是实时统计一段时间内的热门 url。例如，需要统计最近10 秒钟内最热门的两个 url 链接，并且每 5 秒钟更新一次。我们知道，这可以用一个滑动窗口来实现，而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集 url 的访问数据，按照不同的 url 进行统计，而后汇总排序并最终输出前两名。这其实就是著名的“Top N”问题。
很显然，简单的增量聚合可以得到 url 链接的访问量，但是后续的排序输出 Top N 就很难实现了。所以接下来我们用窗口处理函数进行实现。
使用 ProcessAllWindowFunction
一种最简单的想法是，我们干脆不区分 url 链接，而是将所有访问数据都收集起来，统一进行统计计算。所以可以不做 keyBy，直接基于 DataStream 开窗，然后使用全窗口函数ProcessAllWindowFunction 来进行处理。
在窗口中可以用一个 HashMap 来保存每个 url 的访问次数，只要遍历窗口中的所有数据，自然就能得到所有 url 的热门度。最后把 HashMap 转成一个列表 ArrayList，然后进行排序、取出前两名输出就可以了。
代码具体实现如下：
package flink.demo; import flink.demo.PoJo.Event; import flink.demo.Source.ClickSource; import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction; import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.util.Collector; import java.sql.Timestamp; import java.util.ArrayList; import java.util.Comparator; import java.util.HashMap; 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 来表示，它包202
含了 url、浏览量（count）以及窗口的起始结束时间。之后对 UrlViewCount 的处理，可以先按窗口分区，然后用 KeyedProcessFunction 来实现。
总结处理流程如下：
（1）读取数据源；
（2）筛选浏览行为（pv）；
（3）提取时间戳并生成水位线；
（4）按照 url 进行 keyBy 分区操作；
（5）开长度为 1 小时、步长为 5 分钟的事件时间滑动窗口；
（6）使用增量聚合函数 AggregateFunction，并结合全窗口函数 WindowFunction 进行窗口聚合，得到每个 url、在每个统计窗口内的浏览量，包装成 UrlViewCount；
（7）按照窗口进行 keyBy 分区操作；
（8）对同一窗口的统计结果数据，使用 KeyedProcessFunction 进行收集并排序输出。糟糕的是，这里又会带来另一个问题。最后我们用 KeyedProcessFunction 来收集数据做排序，这时面对的就是窗口聚合之后的数据流，而窗口已经不存在了；那到底什么时候会收集齐所有数据呢？这问题听起来似乎有些没道理。我们统计浏览量的窗口已经关闭，就说明了当前
已经到了要输出结果的时候，直接输出不就行了吗？
没有这么简单。因为数据流中的元素是逐个到来的，所以即使理论上我们应该“同时”收到很多 url 的浏览量统计结果，实际也是有先后的、只能一条一条处理。下游任务（就是我们定义的 KeyedProcessFunction）看到一个 url 的统计结果，并不能保证这个时间段的统计数据不会再来了，所以也不能贸然进行排序输出。解决的办法，自然就是要等所有数据到齐了——这很容易让我们联想起水位线设置延迟时间的方法。这里我们也可以“多等一会儿”，等到水位线真正超过了窗口结束时间，要统计的数据就肯定到齐了。
具体实现上，可以采用一个延迟触发的事件时间定时器。基于窗口的结束时间来设定延迟，其实并不需要等太久——因为我们是靠水位线的推进来触发定时器，而水位线的含义就是“之前的数据都到齐了”。所以我们只需要设置 1 毫秒的延迟，就一定可以保证这一点。
而在等待过程中，之前已经到达的数据应该缓存起来，我们这里用一个自定义的“列表状态”（ListState）来进行存储，如图 7-2 所示。这个状态需要使用富函数类的 getRuntimeContext()方法获取运行时上下文来定义，我们一般把它放在 open()生命周期方法中。之后每来一个UrlViewCount，就把它添加到当前的列表状态中，并注册一个触发时间为窗口结束时间加 1毫秒（windowEnd + 1）的定时器。待到水位线到达这个时间，定时器触发，我们可以保证当前窗口所有 url 的统计结果 UrlViewCount 都到齐了；于是从状态中取出进行排序输出。
图 7-2 使用“列表状态”进行排序
具体代码实现如下：
import flink.demo.PoJo.Event; import flink.demo.PoJo.UrlViewCount; import flink.demo.Source.ClickSource; import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.AggregateFunction; import org.apache.flink.api.common.state.ListState; import org.apache.flink.api.common.state.ListStateDescriptor; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.util.Collector; import java.sql.Timestamp; import java.util.ArrayList; import java.util.Comparator; 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 方法读取列表状态变量中的所有元素。
另外，根据水位线的定义，我们这里的延迟时间设为 0 事实上也是可以保证数据都到齐的。感兴趣的读者可以自行修改代码进行测试。
