Flink 中的处理函数-第七章

Posted 2023-04-02 王雀跃

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Flink 中的处理函数-第七章相关的知识，希望对你有一定的参考价值。

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

Flink 中的处理函数

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

在更底层，我们可以不定义任何具体的算子（如 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 。

具体代码实现如下：

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)

Flink处理函数实战之五：CoProcessFunction(双流处理)

欢迎访问我的GitHub

本篇概览

本文是《Flink处理函数实战》系列的第五篇，学习内容是如何同时处理两个数据源的数据；
试想在面对两个输入流时，如果这两个流的数据之间有业务关系，该如何编码实现呢，例如下图中的操作，同时监听9998和9999端口，将收到的输出分别处理后，再由同一个sink处理(打印)：
Flink支持的方式是扩展CoProcessFunction来处理，为了更清楚认识，我们把KeyedProcessFunction和CoProcessFunction的类图摆在一起看，如下所示：
从上图可见，CoProcessFunction和KeyedProcessFunction的继承关系一样，另外CoProcessFunction自身也很简单，在processElement1和processElement2中分别处理两个上游流入的数据即可，并且也支持定时器设置；

编码实战

接下来咱们开发一个应用来体验CoProcessFunction，功能非常简单，描述如下：
1. 建两个数据源，数据分别来自本地9998和9999端口；
2. 每个端口收到类似aaa,123这样的数据，转成Tuple2实例，f0是aaa，f1是123；
3. 在CoProcessFunction的实现类中，对每个数据源的数据都打日志，然后全部传到下游算子；
4. 下游操作是打印，因此9998和9999端口收到的所有数据都会在控制台打印出来；
5. 整个demo的功能如下图所示：
接下来编码实现上述功能；

源码下载

如果您不想写代码，整个系列的源码可在GitHub下载到，地址和链接信息如下表所示(https://github.com/zq2599/blog_demos)：

名称	链接	备注
项目主页	https://github.com/zq2599/blog_demos	该项目在GitHub上的主页
git仓库地址(https)	https://github.com/zq2599/blog_demos.git	该项目源码的仓库地址，https协议
git仓库地址(ssh)	git@github.com:zq2599/blog_demos.git	该项目源码的仓库地址，ssh协议

这个git项目中有多个文件夹，本章的应用在flinkstudy文件夹下，如下图红框所示：

Map算子
1. 做一个map算子，用来将字符串aaa,123转成Tuple2实例，f0是aaa，f1是123；
2. 算子名为WordCountMap.java：
```
package com.bolingcavalry.coprocessfunction;
```

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.StringUtils;

public class WordCountMap implements MapFunction<String, Tuple2<String, Integer>> br/>@Override
public Tuple2<String, Integer> map(String s) throws Exception

    if(StringUtils.isNullOrWhitespaceOnly(s)) 
        System.out.println("invalid line");
        return null;
    

    String[] array = s.split(",");

    if(null==array || array.length<2) 
        System.out.println("invalid line for array");
        return null;
    

    return new Tuple2<>(array[0], Integer.valueOf(array[1]));

### 便于扩展的抽象类
- 开发一个抽象类，将前面图中提到的监听端口、map处理、keyby处理、打印都做到这个抽象类中，但是CoProcessFunction的逻辑却不放在这里，而是交给子类来实现，这样如果我们想进一步实践和扩展CoProcessFunction的能力，只要在子类中专注做好CoProcessFunction相关开发即可，如下图，红色部分交给子类实现，其余的都是抽象类完成的：
![在这里插入图片描述](https://s4.51cto.com/images/blog/202204/06175412_624d63442c57033416.png?x-oss-process=image/watermark,size_14,text_QDUxQ1RP5Y2a5a6i,color_FFFFFF,t_100,g_se,x_10,y_10,shadow_20,type_ZmFuZ3poZW5naGVpdGk=)
- 抽象类AbstractCoProcessFunctionExecutor.java，源码如下，稍后会说明几个关键点：
```java
package com.bolingcavalry.coprocessfunction;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;

/**
 * @author will
 * @email zq2599@gmail.com
 * @date 2020-11-09 17:33
 * @description 串起整个逻辑的执行类，用于体验CoProcessFunction
 */
public abstract class AbstractCoProcessFunctionExecutor 

    /**
     * 返回CoProcessFunction的实例，这个方法留给子类实现
     * @return
     */
    protected abstract CoProcessFunction<
            Tuple2<String, Integer>,
            Tuple2<String, Integer>,
            Tuple2<String, Integer>> getCoProcessFunctionInstance();

    /**
     * 监听根据指定的端口，
     * 得到的数据先通过map转为Tuple2实例，
     * 给元素加入时间戳，
     * 再按f0字段分区，
     * 将分区后的KeyedStream返回
     * @param port
     * @return
     */
    protected KeyedStream<Tuple2<String, Integer>, Tuple> buildStreamFromSocket(StreamExecutionEnvironment env, int port) 
        return env
                // 监听端口
                .socketTextStream("localhost", port)
                // 得到的字符串"aaa,3"转成Tuple2实例，f0="aaa"，f1=3
                .map(new WordCountMap())
                // 将单词作为key分区
                .keyBy(0);
    

    /**
     * 如果子类有侧输出需要处理，请重写此方法，会在主流程执行完毕后被调用
     */
    protected void doSideOutput(SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream) 
    

    /**
     * 执行业务的方法
     * @throws Exception
     */
    public void execute() throws Exception 
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 并行度1
        env.setParallelism(1);

        // 监听9998端口的输入
        KeyedStream<Tuple2<String, Integer>, Tuple> stream1 = buildStreamFromSocket(env, 9998);

        // 监听9999端口的输入
        KeyedStream<Tuple2<String, Integer>, Tuple> stream2 = buildStreamFromSocket(env, 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream = stream1
                // 两个流连接
                .connect(stream2)
                // 执行低阶处理函数，具体处理逻辑在子类中实现
                .process(getCoProcessFunctionInstance());

        // 将低阶处理函数输出的元素全部打印出来
        mainDataStream.print();

        // 侧输出相关逻辑，子类有侧输出需求时重写此方法
        doSideOutput(mainDataStream);

        // 执行
        env.execute("ProcessFunction demo : CoProcessFunction");

关键点之一：一共有两个数据源，每个源的处理逻辑都封装到buildStreamFromSocket方法中；
关键点之二：stream1.connect(stream2) 将两个流连接起来；
关键点之三：process接收CoProcessFunction实例，合并后的流的处理逻辑就在这里面；
关键点之四：getCoProcessFunctionInstance是抽象方法，返回CoProcessFunction实例，交给子类实现，所以CoProcessFunction中做什么事情完全由子类决定；
关键点之五：doSideOutput方法中啥也没做，但是在主流程代码的末尾会被调用，如果子类有侧输出(SideOutput)的需求，重写此方法即可，此方法的入参是处理过的数据集，可以从这里取得侧输出；

子类决定CoProcessFunction的功能

子类CollectEveryOne.java如下所示，逻辑很简单，将每个源的上游数据直接输出到下游算子：
```
package com.bolingcavalry.coprocessfunction;
```

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CollectEveryOne extends AbstractCoProcessFunctionExecutor

private static final Logger logger = LoggerFactory.getLogger(CollectEveryOne.class);

@Override
protected CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> getCoProcessFunctionInstance() 
    return new CoProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>>() 

        @Override
        public void processElement1(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) 
            logger.info("处理1号流的元素：,", value);
            out.collect(value);
        

        @Override
        public void processElement2(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) 
            logger.info("处理2号流的元素：", value);
            out.collect(value);
        
    ;


public static void main(String[] args) throws Exception 
    new CollectEveryOne().execute();

2. 上述代码中，CoProcessFunction后面的泛型定义很长：<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> ，一共三个Tuple2，分别代表一号数据源输入、二号数据源输入、下游输出的类型；
### 验证
1. 分别开启本机的**9998**和**9999**端口，我这里是MacBook，执行**nc -l 9998**和**nc -l 9999**
2. 启动Flink应用，如果您和我一样是Mac电脑，直接运行**CollectEveryOne.main**方法即可（如果是windows电脑，我这没试过，不过做成jar在线部署也是可以的）；
3. 在监听9998和9999端口的控制台分别输入**aaa,111**和**bbb,222**
4. 以下是flink控制台输出的内容，可见processElement1和processElement1方法的日志代码已经执行，并且print方法作为最下游，将两个数据源的数据都打印出来了，符合预期：
```shell
12:45:38,774 INFO CollectEveryOne - 处理1号流的元素：(aaa,111),
(aaa,111)
12:45:43,816 INFO CollectEveryOne - 处理2号流的元素：(bbb,222)
(bbb,222)