Flink 中的处理函数-第七章

小灰灰 2023-10-08 23:22 26阅读 0赞

> 借鉴《尚硅谷Flink1.13版本笔记.pdf》中第七章

## Flink 中的处理函数 ##

之前所介绍的流处理 API，无论是基本的转换、聚合，还是更为复杂的窗口操作，都是基于 DataStream 进行转换；所以可以统称为 DataStream API，这是 Flink 编程的核心。 而我们知道，为让代码有更强大的表现力和易用性，Flink 本身提供了多层 API，DataStream API 只是中间的一环，如图 7-1 ：

![2cd470e533924959bc2a873d9beccd22.png][]在更底层，我们可以不定义任何具体的算子（如 map()，filter()，或者 window()），而只提炼出一个统一的“处理”（process）操作——它是所有转换算子的一个概括性的表达，可以自定义处理逻辑，所以这一层接口就被叫作“处理函数”（process function）。

--------------------

### 7.1 基本处理函数（ProcessFunction） ###

处理函数主要是定义数据流的转换操作，也可以把它归到转换算子中。我们知道在 Flink 中几乎所有转换算子都提供了对应的函数类接口，处理函数也不例外；它所对应的函数类，就叫作 ProcessFunction。

#### 7.1.1 处理函数的功能和使用 ####

之前学习的转换算子，只是针对某种具体操作来定义，能够拿到的信息有限。

如 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）， 富函数可以调用的这里同样都可以调用。 下面是一个具体的应用示例：

import com.atguigu.chapter05.{ClickSource, Event}
    import org.apache.flink.streaming.api.functions._
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    
    object ProcessFunctionExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     env
     .addSource(new ClickSource)
     .assignAscendingTimestamps(_.timestamp)
     .process(new ProcessFunction[Event, String] {
     // 每来一条元素都会调用一次
     override def processElement(i: Event, context: ProcessFunction[Event, 
    String]#Context, collector: Collector[String]): Unit = {
     if (i.user.equals("Mary")) {
     // 向下游发送数据
     collector.collect(i.user)
    119
     } else if (i.user.equals("Bob")) {
     collector.collect(i.user)
     collector.collect(i.user)
     }
     // 打印当前水位线
     println(context.timerService.currentWatermark())
     }
     })
     .print()
     env.execute()
     }}

--------------------

#### 7.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()。 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. 非抽象方法 onTimer()***

该方法用于定义定时触发的操作，这是一个强大、有趣的功能。这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过“定时服务”TimerService 来注册的。 打个比方，注册定时器（timer）就是设了一个闹钟，到了设定时间就会响；而 onTimer()中定义的，就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”（callback）方法， 通过时间的进展来触发；在事件时间语义下就是由水位线（watermark）来触发了。

与 processElement()类似，定时方法 onTimer()也有三个参数：时间戳（timestamp），上下文（ctx），收集器（out）。

这里 timestamp 是指设定好的触发时间，事件时间语义下就是水位线。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService）， 以及任意输出处理之后的数据。

既然有.onTimer()方法做定时触发，我们用 ProcessFunction 也可以自定义数据按照时间分组、定时触发计算输出结果；这其实就实现了窗口（window）的功能。

这里需要注意，上面的 onTimer()方法只是定时器触发时的操作，而定时器（timer） 真正的设置需要用到上下文 ctx 中的定时服务。在 Flink 中，只有“按键分区流”KeyedStream 才支持设置定时器的操作，所以之前的代码中并没有用定时器。所以基于不同类型的流， 可以使用不同的处理函数，它们之间有些微小区别的。接下来我们就介绍一下处理函数的分类。

--------------------

***7.1.3 处理函数的分类***

Flink 中处理函数是一个大家族，ProcessFunction 只是其中一员。 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 的具体用法详细说明。

--------------------

### 7.2 按键分区处理函数（KeyedProcessFunction） ###

在 Flink 程序中，为实现数据的聚合统计，或者开窗计算之类的功能，一般都要先用 keyBy()算子对数据流进行“按键分区”，得到 KeyedStream。而只有在 KeyedStream 中， 才支持使用 TimerService 设置定时器的操作。

所以一般情况下，我们都是先做了 keyBy()分区后，再去定义处理操作；代码中更加常见的处理函数是 KeyedProcessFunction。 接下来我们就先从定时服务（TimerService）入手，讲解 KeyedProcessFunction 的用法

#### 7.2.1 定时器（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 不支持定时器操作，只能获取当前时间。

可以认为，定时器其实是 KeyedStream 上处理算子的一个状态，它以时间戳作为区分。所以 TimerService 会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个 key 和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。

--------------------

#### 7.2.2 KeyedProcessFunction 的使用 ####

代码中使用 KeyedProcessFunction，只要基于 keyBy()之后的 KeyedStream，直接调用 process()方法，这时需要传入的参数就是 KeyedProcessFunction 的实现类。

stream.keyBy( _._1 )
    .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 的类型。同样地，我们必须实现一个 processElement()抽象方法，来处理流中的每一个数据；另外还有一个非抽象方法 onTimer()，用来定义定时器触发时的回调操作。 下面是一个使用处理时间定时器的具体示例：

import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.util.Collector
    import java.sql.Timestamp
    
    object ProcessingTimeTimerExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     env
     .addSource(new ClickSource)
     .keyBy(r => true)
     .process(new KeyedProcessFunction[Boolean, Event, String] {
     override def processElement(value: Event, ctx: 
    KeyedProcessFunction[Boolean, Event, String]#Context, out: Collector[String]): 
    Unit = {
     val currTs = ctx.timerService.currentProcessingTime()
     out.collect("数据到达，到达时间：" + new Timestamp(currTs))
     // 注册 10 秒钟之后的处理时间定时器
    124
     ctx.timerService().registerProcessingTimeTimer(currTs + 10 * 1000L)
     }
     // 定时器的逻辑
     override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Boolean, 
    Event, String]#OnTimerContext, out: Collector[String]): Unit = {
     out.collect("定时器触发，触发时间：" + new Timestamp(timestamp))
     }
     })
     .print()
     env.execute()
     }
    }

上面的代码中，由于定时器只能在 KeyedStream 上使用，所以先要 keyBy()；这里 的 keyBy(r -> true)是将所有数据的 key 都指定为 true，就是所有数据拥有相同的 key， 会分配到同一个分区。

--------------------

### 7.3 窗口处理函数 ###

除 KeyedProcessFunction ， 另外一大类常用的处理函数，就是基于窗口的 ProcessWindowFunction 和 ProcessAllWindowFunction 。

#### 7.3.1 窗口处理函数的使用 ####

进行窗口计算，可以直接调用现成的简单聚合方法（sum()/max()/min()）,也可以通过调用 reduce()或 aggregate()来自定义一般的增量聚合函数（ReduceFunction/AggregateFucntion）； 而对于更复杂、需要窗口信息和额外状态的一些场景，我们还可以直接使用全窗口函数、把数据全部收集保存在窗口内，等触发窗口计算时再统一处理。窗口处理函数就是一种典型的全窗口函数。

窗口处理函数 ProcessWindowFunction 的使用与其他窗口函数类似 ， 也是基于 WindowedStream 直接调用方法就可以，只不过这时调用的是 process()。

stream.keyBy(_._1)
     .window( TumblingEventTimeWindows.of(Time.seconds(10)) )
     .process(new MyProcessWindowFunction)

--------------------

#### 7.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。 而 ProcessWindowFunction 内部定义的方法，跟我们之前熟悉的处理函数就有所区别了。 因为全窗口函数不是逐个处理元素的，所以处理数据的方法在这里并不是 processElement()， 而是改成 process()。方法包含四个参数。

⚫ key：窗口做统计计算基于的键，也就是之前 keyBy()用来分区的字段。

⚫ context：当前窗口进行计算的上下文，它的类型就是 ProcessWindowFunction 内部定义的抽象类 Context。

⚫ elements：窗口收集到用来计算的所有数据，这是一个可迭代的集合类型。

⚫ out：用来发送数据输出计算结果的收集器，类型为 Collector。 可明显看出，这里的参数 elements 不再是一个输入数据，而是窗口中所有数据的集合。 ProcessWindowFunction 中除.process()方法外，并没有 onTimer()方法，而是多出了一个 clear()方法，这主要是方便我们进行窗口的清理工作。

至于另一种窗口处理函数 ProcessAllWindowFunction，它的用法非常类似。区别在于它基于的是 AllWindowedStream，相当于对没有 keyBy()的数据流直接开窗并调用 process()方法:

stream.windowAll( TumblingEventTimeWindows.of(Time.seconds(10)) )
    .process(new MyProcessAllWindowFunction)

--------------------

### 7.4 应用案例——Top N ###

窗口的计算处理，在实际应用中非常常见。对一些复杂的需求，如果增量聚合函数无法满足，就需要考虑使用窗口处理函数。

网站中一个非常经典的例子，就是实时统计一段时间内的热门 url。例如，需要统计最近 10 秒钟内最热门的两个 url 链接，且每 5 秒钟更新一次。我们知道，这可以用一个滑动窗口实现，而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集 url 的访问数据，按照不同的 url 进行统计，而后汇总排序并最终输出前两名。这就是“Top N” 问题。

很显然，简单的增量聚合可以得到 url 链接的访问量，但后续的排序输出 Top N 很难实现。所以接下来用窗口处理函数进行实现。

#### 7.4.1 使用 ProcessAllWindowFunction ####

一种最简单的想法是，干脆不区分 url 链接，而是将所有访问数据都收集起来，统一进行统计计算。所以可以不做 keyBy，直接基于 DataStream 开窗，然后使用全窗口函数 ProcessAllWindowFunction 来进行处理。

在窗口中用一个 HashMap 来保存每个 url 的访问次数，只要遍历窗口中的所有数据， 自然就能得到所有 url 的热门度。最后把 HashMap 转成一个列表 ArrayList，进行排序、 取出前两名输出了。 代码具体实现如下：

object ProcessAllWindowTopNExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     val eventStream = env
     .addSource(new ClickSource)
     .assignAscendingTimestamps(_.timestamp)
     // 只需要 url 就可以统计数量，所以抽取 url 转换成 String，直接开窗统计
     eventStream.map(_.url)
     // 开窗口
     .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
     .process(new ProcessAllWindowFunction[String, String, TimeWindow] {
     override def process(context: Context, elements: Iterable[String], out: 
    Collector[String]): Unit = {
    127
     // 初始化一个 Map，key 为 url，value 为 url 的 pv 数据
     val urlCountMap = Map[String, Long]()
     // 将 url 和 pv 数据写入 Map 中
     elements.foreach(
     r => urlCountMap.get(r) match {
     case Some(count) => urlCountMap.put(r, count + 1L)
     case None => urlCountMap.put(r, 1L)
     }
     )
     // 将 Map 中的 KV 键值对转换成列表数据结构
     // 列表中的元素是(K,V)元组
     var mapList = new ListBuffer[(String, Long)]()
     urlCountMap.keys.foreach(
     k => urlCountMap.get(k) match {
     case Some(count) => mapList += ((k, count))
     case None => mapList
     }
     )
     // 按照浏览量数据进行降序排列
     mapList.sortBy(-_._2)
     // 拼接字符串并输出
     val result = new StringBuilder
     result.append("==================================\n")
     for (i <- 0 to 1) {
     val temp = mapList(i)
     result
     .append("浏览量 No." + (i + 1) + " ")
     .append("url: " + temp._1 + " ")
     .append("浏览量是：" + temp._2 + " ")
     .append("窗口结束时间是：" + new Timestamp(context.window.getEnd) + 
    "\n")
     }
     result.append("===================================\n")
     out.collect(result.toString())
     }
     })
     .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
    ========================================

--------------------

#### 7.4.2 使用 KeyedProcessFunction ####

在上一小节的实现过程中，没有进行按键分区，直接将所有数据放在一个分区上进行开窗操作。这相当于将并行度强行设置为 1，在实际应用中是要避免，所以 Flink 官方不推荐使用 AllWindowedStream 进行处理。

如果可以利用增量聚合函数的特性，每来一条数据就更新一次对应 url 的浏览量，那么到窗口触发计算时只需要做排序输出就可以了。

基于这样的想法，我们可以从两个方面去做优化：一是对数据进行按键分区，分别统计浏览量；

二是进行增量聚合，得到结果最后再做排序输出。

所以，我们可以使用增量聚合函数 AggregateFunction 进行浏览量的统计，然后结合 ProcessWindowFunction 排序输出来实现 TopN 需求。

具体实现，可以分成两步：先对每个 url 链接统计出浏览量，再将统计结果收集起来，排序输出最终结果。 而为了将同一窗口的所有 url 统计结果收集齐，需要设置一个延迟触发的事件时间定时器，来进行等待。我们只需要基于窗口结束时间设置 1 毫秒的延迟，就可以保证所有数据都已到齐了。 而在等待过程中，之前已经到达的数据应该缓存起来，这里用一个自定义的“列表状态”（ListState）来进行存储，如图 7-2 。

![e119939f264948b990106ff67f513ff9.png][]

具体代码实现如下：

import org.apache.flink.api.common.functions.AggregateFunction
    import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
    import org.apache.flink.streaming.api.functions.KeyedProcessFunction
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.streaming.api.scala.function.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 com.atguigu.chapter05.{ClickSource, Event}
    import org.apache.flink.configuration.Configuration
    
    object KeyedProcessTopNExample {
     def main(args: Array[String]): Unit = {
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(1)
     val eventStream = env
     .addSource(new ClickSource)
     .assignAscendingTimestamps(_.timestamp)
     // 需要按照 url 分组，求出每个 url 的访问量
     val urlCountStream = eventStream
     .keyBy(_.url)
     // 开窗口
     .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
     // 增量聚合函数和全窗口聚合函数结合使用
     // 计算结果是每个窗口中每个 url 的浏览次数
     .aggregate(new UrlViewCountAgg, new UrlViewCountResult)
     // 对结果中同一个窗口的统计数据，进行排序处理
     val result = urlCountStream
     .keyBy(_.windowEnd)
     .process(new TopN(2))
     result.print()
     env.execute()
     }
     class TopN(n: Int) extends KeyedProcessFunction[Long, UrlViewCount, String] {
     // 定义列表状态，存储 UrlViewCount 数据
     var urlViewCountListState: ListState[UrlViewCount] = _
     override def open(parameters: Configuration): Unit = {
     urlViewCountListState = getRuntimeContext.getListState(
     new ListStateDescriptor[UrlViewCount]("list-state", 
    classOf[UrlViewCount]))
     }
     override def processElement(i: UrlViewCount, context: 
    KeyedProcessFunction[Long, UrlViewCount, String]#Context, collector: 
    Collector[String]): Unit = {
     // 每来一条数据就添加到列表状态变量中
     urlViewCountListState.add(i)
     // 注册一个定时器，由于来的数据的 windowEnd 是相同的，所以只会注册一个定时器
     context.timerService.registerEventTimeTimer(i.windowEnd + 1)
     }
     override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, 
    UrlViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {
     // 导入隐式类型转换
     import scala.collection.JavaConversions._
     // 下面的代码将列表状态变量里的元素取出，然后放入 List 中，方便排序
     val urlViewCountList = urlViewCountListState.get().toList
     // 由于数据已经放入 List 中，所以可以将状态变量手动清空了
     urlViewCountListState.clear()
     // 按照浏览次数降序排列
     urlViewCountList.sortBy(-_.count)
     // 拼接要输出的字符串
     val result = new StringBuilder
     result.append("=========================\n")
     for (i <- 0 until n) {
     val urlViewCount = urlViewCountList(i)
     result
     .append("浏览量 No." + (i + 1) + " ")
     .append("url: " + urlViewCount.url + " ")
     .append("浏览量：" + urlViewCount.count + " ")
     .append("窗口结束时间：" + new Timestamp(timestamp - 1L) + "\n")
     }
     result.append("=========================\n")
     out.collect(result.toString())
     }
     }
     class UrlViewCountAgg extends AggregateFunction[Event, Long, Long] {
     override def createAccumulator(): Long = 0L
     override def add(value: Event, accumulator: Long): Long = accumulator + 1L
     override def getResult(accumulator: Long): Long = accumulator
     override def merge(a: Long, b: Long): Long = ???
     }
     class UrlViewCountResult extends ProcessWindowFunction[Long, UrlViewCount, 
    String, TimeWindow] {
     override def process(key: String, context: Context, elements: Iterable[Long], 
    out: Collector[UrlViewCount]): Unit = {
     // 迭代器中只有一条元素，就是增量聚合函数发送过来的聚合结果
     out.collect(UrlViewCount(
     key, elements.iterator.next(), context.window.getStart, 
    context.window.getEnd
     ))
     }
     }
     case class UrlViewCount(url: String, count: Long, windowStart: Long, windowEnd: 
    Long)
    }

--------------------

### 7.5 侧输出流（Side Output） ###

处理函数还有另外一个特有功能，就是将自定义的数据放入“侧输出流”（side output）输出。这个概念并不陌生，之前在讲到窗口处理迟到数据时，最后一招就是输出到侧输出流。 而这种处理方式的本质，就是处理函数的侧输出流功能。 具体应用时，只要在处理函数的 processElement()或者 onTimer()方法中，调用上下文的 output()方法就可以了。

val stream = env.addSource(new ClickSource)
    val longStream = stream.process(new ProcessFunction[Event, Long] {
     override def processElement(value: Event, ctx: ProcessFunction[Event, 
    Long]#Context, out: Collector[Long]) = {
     //将时间戳输出到主流中
     out.collect(value.timestamp)
     //将用户名输出到侧输出流中
     ctx.output(outputTag, "side-output: " + value.user)
     }
     })

这里 output()方法需传入两个参数，第一个是一个“输出标签”OutputTag，用来标识侧输出流，一般会在外部统一声明；第二个就是要输出的数据。 可以在外部先将 OutputTag 声明出来：

val outputTag: OutputTag[String] = OutputTag[String]("user")

如果要获取这个侧输出流，可基于处理之后的 DataStream 直接调用 getSideOutput() 方法，传入对应的 OutputTag，这个方式与窗口 API 中获取侧输出流一样。

val stringStream = longStream.getSideOutput(outputTag)

[2cd470e533924959bc2a873d9beccd22.png]: https://img-blog.csdnimg.cn/2cd470e533924959bc2a873d9beccd22.png
[e119939f264948b990106ff67f513ff9.png]: https://img-blog.csdnimg.cn/e119939f264948b990106ff67f513ff9.png