Flink处理函数（一）

Flink处理函数（一）
目录

7.1 基本处理函数（ProcessFunction）

7.1.1 处理函数的功能和使用

7.1.2 ProcessFunction 解析

7.1.3 处理函数的分类

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

7.2.1 定时器（Timer）和定时服务（TimerService）

7.2.2 KeyedProcessFunction 的使用

7.3 窗口处理函数

7.3.1 窗口处理函数的使用

7.3.2 ProcessWindowFunction 解析

7.4 应用案例——Top N

7.4.1 使用 ProcessAllWindowFunction

7.4.2 使用 KeyedProcessFunction

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

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

在处理函数中，我们直面的就是数据流中最基本的元素：数据事件（event）、状态（state）以及时间（time）。这就相当于对流有了完全的控制权。处理函数比较抽象，没有具体的操作，所以对于一些常见的简单应用（比如求和、开窗口）会显得有些麻烦；不过正是因为它不限定具体做什么，所以理论上我们可以做任何事情，实现所有需求。所以可以说，处理函数是我们进行 Flink 编程的“大招”，轻易不用，一旦放出来必然会扫平一切。

7.1 基本处理函数（ProcessFunction）

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

7.1.1 处理函数的功能和使用

我们之前学习的转换算子，一般只是针对某种具体操作来定义的，能够拿到的信息比较有限。比如 map 算子，我们实现的 MapFunction 中，只能获取到当前的数据，定义它转换之后的形式；而像窗口聚合这样的复杂操作，AggregateFunction 中除数据外，还可以获取到当前的状态（以累加器 Accumulator 形式出现）。另外我们还介绍过富函数类，比如 RichMapFunction，它提供了获取运行时上下文的方法 getRuntimeContext()，可以拿到状态，还有并行度、任务名称之类的运行时信息。

但是无论那种算子，如果我们想要访问事件的时间戳，或者当前的水位线信息，都是完全做不到的。处理函数提供了一个“定时服务” （TimerService），我们可以通过它访问流中的事件（event）、时间戳（timestamp）、水位线（watermark），甚至可以注册“定时事件”。而且处理函数继承了 AbstractRichFunction 抽象类，所以拥有富函数类的所有特性，同样可以访问状态（state）和其他运行时信息。此外，处理函数还可以直接将数据输出到侧输出流（side output）中。所以，处理函数是最为灵活的处理方法，可以实现各种自定义的业务逻辑；同时也是整个 DataStream API 的底层基础。

处理函数的使用与基本的转换操作类似，只需要直接基于 DataStream 调用.process()方法就可以了。方法需要传入一个 ProcessFunction 作为参数，用来定义处理逻辑。
```
stream.process(new MyProcessFunction()) 
```
这里 ProcessFunction 不是接口，而是一个抽象类，继承了 AbstractRichFunction；

MyProcessFunction 是它的一个具体实现。所以所有的处理函数，都是富函数（RichFunction）， 富函数可以调用的东西这里同样都可以调用。

7.1.2 ProcessFunction 解析

抽象类 ProcessFunction 继承了 AbstractRichFunction，有两个泛型类型参数：I 表示 Input，也就是输入的数据类型；O 表示 Output，也就是处理完成之后输出的数据类型。

内部单独定义了两个方法：一个是必须要实现的抽象方法.processElement()；另一个是非抽象方法.onTimer()。
```
public abstract class ProcessFunction extends AbstractRichFunction { 
 ... 
public abstract void processElement(I value, Context ctx, Collector out) 
throws Exception; 
public void onTimer(long timestamp, OnTimerContext ctx, Collector 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  void output(OutputTag 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）的功能。所以说 ProcessFunction是真正意义上的终极奥义，用它可以实现一切功能。

我们也可以看到，处理函数都是基于事件触发的。水位线就如同插入流中的一条数据一样；只不过处理真正的数据事件调用的是.processElement()方法，而处理水位线事件调用的是.onTimer()。

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

7.1.3 处理函数的分类

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）做连接之后的产物。

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

在 Flink 程序中，为了实现数据的聚合统计，或者开窗计算之类的功能，我们一般都要先用 keyBy 算子对数据流进行“按键分区”，得到一个 KeyedStream。也就是指定一个键（key），按照它的哈希值（hash code）将数据分成不同的“组”，然后分配到不同的并行子任务上执行计算；这相当于做了一个逻辑分流的操作，从而可以充分利用并行计算的优势实时处理海量数据。

另外我们在上节中也提到，只有在 KeyedStream 中才支持使用 TimerService 设置定时器的操作。所以一般情况下，我们都是先做了 keyBy 分区之后，再去定义处理操作；代码中更加常见的处理函数是 KeyedProcessFunction，最基本的 ProcessFunction 反而出镜率没那么高。

接下来我们就先从定时服务（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 不支持定时器操作，只能获取当前时间。

对于处理时间和事件时间这两种类型的定时器，TimerService 内部会用一个优先队列将它们的时间戳（timestamp）保存起来，排队等待执行。可以认为，定时器其实是 KeyedStream

上处理算子的一个状态，它以时间戳作为区分。所以 TimerService 会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个 key 和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。这样一来，我们在代码中就方便了很多，可以肆无忌惮地对一个 key 注册定时器，而不用担心重复定义——因为一个时间戳上的定时器只会触发一次。

基于 KeyedStream 注册定时器时，会传入一个定时器触发的时间戳，这个时间戳的定时器对于每个 key 都是有效的。这样，我们的代码并不需要做额外的处理，底层就可以直接对不同key 进行独立的处理操作了。

利用这个特性，有时我们可以故意降低时间戳的精度，来减少定时器的数量，从而提高处理性能。比如我们可以在设置定时器时只保留整秒数，那么定时器的触发频率就是最多 1 秒一次。
```
long coalescedTime = time / 1000 * 1000; 
ctx.timerService().registerProcessingTimeTimer(coalescedTime); 
```
这里注意定时器的时间戳必须是毫秒数，所以我们得到整秒之后还要乘以 1000。定时器默认的区分精度是毫秒。

另外 Flink 对.onTimer()和.processElement()方法是同步调用的（synchronous），所以也不会出现状态的并发修改。 Flink 的定时器同样具有容错性，它和状态一起都会被保存到一致性检查点（checkpoint）中。当发生故障时，Flink 会重启并读取检查点中的状态，恢复定时器。如果是处理时间的定时器，有可能会出现已经“过期”的情况，这时它们会在重启时被立刻触发。

7.2.2 KeyedProcessFunction 的使用

KeyedProcessFunction 可以说是处理函数中的“嫡系部队”，可以认为是 ProcessFunction 的一个扩展。我们只要基于 keyBy 之后的 KeyedStream，直接调用.process()方法，这时需要传入的参数就是 KeyedProcessFunction 的实现类。
```
stream.keyBy( t -> t.f0 ) 
.process(new MyKeyedProcessFunction()) 
```
类似地，KeyedProcessFunction 也是继承自 AbstractRichFunction 的一个抽象类，源码中定义如下：
```
public abstract class KeyedProcessFunction extends AbstractRichFunction 
{ 
 ... 
public abstract void processElement(I value, Context ctx, Collector out) 
throws Exception; 
public void onTimer(long timestamp, OnTimerContext ctx, Collector out) 
throws Exception {} 
public abstract class Context {...} 
... 
} 
```
可以看到与 ProcessFunction 的定义几乎完全一样，区别只是在于类型参数多了一个 K，这是当前按键分区的 key 的类型。同样地，我们必须实现一个.processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法.onTimer()，用来定义定时器触发时的回调操作。由于定时器只能在 KeyedStream 上使用，所以到了 KeyedProcessFunction 这里，我们才真正对时间有了精细的控制，定时方法.onTimer()才真正派上了用场。
```
package com.atguigu.chapter07;
 
import com.atguigu.chapter05.ClickSource;
import com.atguigu.chapter05.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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.util.Collector;
 
import java.sql.Timestamp;
 
public class ProcessingTimeTimer {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        SingleOutputStreamOperator stream = env.addSource(new ClickSource());
 
        stream.keyBy(data -> data.user)
                        .process(new KeyedProcessFunction() {
                            @Override
                            public void processElement(Event event, KeyedProcessFunction.Context context, Collector collector) throws Exception {
                                Long currTs=context.timerService().currentProcessingTime();
                                collector.collect(context.getCurrentKey()+"数据到达，到达时间："+new Timestamp(currTs));
 
                                //注册一个10秒后的定时器
                                context.timerService().registerProcessingTimeTimer(currTs+10*1000L);
                            }
 
                            //定时器
 
                            @Override
                            public void onTimer(long timestamp, KeyedProcessFunction.OnTimerContext ctx, Collector out) throws Exception {
                                out.collect(ctx.getCurrentKey()+"定时器触发，触发时间"+new Timestamp(timestamp));
                            }
                        })
                .print();
 
        env.execute();
    }
}
```
在上面的代码中，由于定时器只能在 KeyedStream 上使用，所以先要进行 keyBy；这里的.keyBy(data -> true)是将所有数据的 key 都指定为了 true，其实就是所有数据拥有相同的 key，会分配到同一个分区。

之后我们自定义了一个 KeyedProcessFunction，其中.processElement()方法是每来一个数据都会调用一次，主要是定义了一个 10 秒之后的定时器；而.onTimer()方法则会在定时器触发时调用。所以我们会看到，程序运行后先在控制台输出“数据到达”的信息，等待 10 秒之后，又会输出“定时器触发”的信息，打印出的时间间隔正是 10 秒。
```
package com.atguigu.chapter07;
 
import com.atguigu.chapter05.ClickSource;
import com.atguigu.chapter05.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.util.Collector;
 
import java.sql.Timestamp;
 
public class EventTimeTimerTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        SingleOutputStreamOperator stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        }));
 
        // 基于 KeyedStream 定义事件时间定时器
        stream.keyBy(data -> true)
                .process(new KeyedProcessFunction() {
                    @Override
                    public void processElement(Event value, Context ctx, Collector 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 out) throws Exception {
                        out.collect(ctx.getCurrentKey()+"定时器触发，触发时间"+new Timestamp(timestamp));
                    }
                })
                .print();
 
        env.execute();
    }
} 
```
7.3 窗口处理函数

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

7.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()) 
```
7.3.2 ProcessWindowFunction 解析

ProcessWindowFunction 既是处理函数又是全窗口函数。从名字上也可以推测出，它的本质似乎更倾向于“窗口函数”一些。事实上它的用法也确实跟其他处理函数有很大不同。我们可以从源码中的定义看到这一点：
```
public abstract class ProcessWindowFunctionextends Window> 
 extends AbstractRichFunction { 
... 
public abstract void process( 
 KEY key, Context context, Iterable elements, Collector 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  void output(OutputTag outputTag, X value); 
} 
```
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，然后进行排序、取出前两名输出就可以了。
```
package com.atguigu.chapter07;
 
import com.atguigu.chapter05.ClickSource;
import com.atguigu.chapter05.Event;
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.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.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.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
 
public class TopExample_ProcessALLWindowFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        //读取数据
        SingleOutputStreamOperator stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                );
 
        //直接开窗，收集所有数据排序
        stream.map(data -> data.url)
                        .windowAll(SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(5)))
                                .aggregate(new URlHashMapCountAgg(),new UrlAllWindowResult())
                                        .print();
 
        env.execute();
    }
    public static class URlHashMapCountAgg implements AggregateFunction, ArrayList>>{
 
        @Override
        public HashMap createAccumulator() {
            return new HashMap();
        }
 
        @Override
        public HashMap add(String value, HashMap stringLongHashMap) {
            /*if(stringLongHashMap.containsKey(value)){
                Long count=stringLongHashMap.get(value);
                stringLongHashMap.put(value,count+1);
            }else {
                stringLongHashMap.put(value,1L);
            }*/
            Long count = stringLongHashMap.getOrDefault(value, 0L);
            stringLongHashMap.put(value,count+1);
            return stringLongHashMap;
        }
 
        @Override
        public ArrayList> getResult(HashMap stringLongHashMap) {
            ArrayList> result = new ArrayList<>();
            for (String key : stringLongHashMap.keySet()) {
                result.add(Tuple2.of(key,stringLongHashMap.get(key)));
            }
            result.sort(new Comparator>() {
                @Override
                public int compare(Tuple2 o1, Tuple2 o2) {
                    return o2.f1.intValue()- o1.f1.intValue();
                }
            });
            return result;
        }
 
        @Override
        public HashMap merge(HashMap stringLongHashMap, HashMap acc1) {
            return null;
        }
    }
    //实现自定义全窗口函数，包装信息输出结果
    public static class UrlAllWindowResult extends ProcessAllWindowFunction>,String, TimeWindow> {
 
 
        @Override
        public void process(ProcessAllWindowFunction>, String, TimeWindow>.Context context, Iterable>> iterable, Collector collector) throws Exception {
            ArrayList> list = iterable.iterator().next();
 
            StringBuilder result = new StringBuilder();
            result.append("-------------------------\n");
            result.append("窗口结束时间："+ new Timestamp(context.window().getEnd()) +"\n");
 
            //取list前两个，包装信息输出
            for(int i=0;i<2;i++){
                Tuple2 currTuple = list.get(i);
                String info="NO."+(i+1)+" "+"url: "+currTuple.f0+" "+"访问量："+currTuple.f1+" \n";
                result.append(info);
            }
            result.append("-------------------------\n");
 
            collector.collect(result.toString());
        }
    }
}
```
7.4.2 使用 KeyedProcessFunction

在上一小节的实现过程中，我们没有进行按键分区，直接将所有数据放在一个分区上进行了开窗操作。这相当于将并行度强行设置为 1，在实际应用中是要尽量避免的，所以 Flink 官方也并不推荐使用 AllWindowedStream 进行处理。另外，我们在全窗口函数中定义了 HashMap来统计 url 链接的浏览量，计算过程是要先收集齐所有数据、然后再逐一遍历更新 HashMap，这显然不够高效。如果我们可以利用增量聚合函数的特性，每来一条数据就更新一次对应 url的浏览量，那么到窗口触发计算时只需要做排序输出就可以了。

基于这样的想法，我们可以从两个方面去做优化：一是对数据进行按键分区，分别统计浏览量；二是进行增量聚合，得到结果最后再做排序输出。所以，我们可以使用增量聚合函数

AggregateFunction 进行浏览量的统计，然后结合 ProcessWindowFunction 排序输出来实现 Top N的需求。

具体实现思路就是，先按照 url 对数据进行 keyBy 分区，然后开窗进行增量聚合。这里就会发现一个问题：我们进行按键分区之后，窗口的计算就会只针对当前 key 有效了；也就是说，每个窗口的统计结果中，只会有一个 url 的浏览量，这是无法直接用 ProcessWindowFunction

进行排序的。所以我们只能分成两步：先对每个 url 链接统计出浏览量，然后再将统计结果收集起来，排序输出最终结果。因为最后的排序还是基于每个时间窗口的，所以为了让输出的统计结果中包含窗口信息，我们可以借用第六章中定义的 POJO 类 UrlViewCount 来表示，它包含了 url、浏览量（count）以及窗口的起始结束时间。之后对 UrlViewCount 的处理，可以先按窗口分区，然后用 KeyedProcessFunction 来实现。
```
package com.atguigu.chapter07;
 
import com.atguigu.chapter05.ClickSource;
import com.atguigu.chapter05.Event;
import com.atguigu.chapter06.UrlCountViewExample;
import com.atguigu.chapter06.UrlViewCount;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
 
import java.sql.Timestamp;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
 
public class TopExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
 
        SingleOutputStreamOperator stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                );
 
        //1.按照url分组，统计窗口内每个url的访问量
        SingleOutputStreamOperator urlCountStream = stream.keyBy(data -> data.url)
                //自定义滑动窗口
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .aggregate(new UrlCountViewExample.UrlViewCountAgg(), new UrlCountViewExample.UrlViewCountResult());
        urlCountStream.print("url count");
 
        //2.对于同一窗口统计出的访问量，进行收集和排序
        urlCountStream.keyBy(data -> data.windowEnd)
                .process(new TopNProcessResult(2))
                .print();
 
        env.execute();
 
    }
 
    //实现自定义的KeyedProcessFunction
    public static class TopNProcessResult extends KeyedProcessFunction{
 
        //定义一个属性n
        private Integer n;
 
        //定义列表状态
        private ListState urlViewCountListState;
 
        public TopNProcessResult(Integer n){
            this.n=n;
        }
 
        //在环境中获取状态
        @Override
        public void open(Configuration parameters) throws Exception {
            urlViewCountListState= getRuntimeContext().getListState(
                    new ListStateDescriptor("url-count-list", Types.POJO(UrlViewCount.class))
            );
        }
 
        @Override
        public void processElement(UrlViewCount value, KeyedProcessFunction.Context context, Collector collector) throws Exception {
            //将数据保存到状态中
            urlViewCountListState.add(value);
            //注册windowEnd +1ms 的定时器
            context.timerService().registerEventTimeTimer(context.getCurrentKey()+1);
        }
 
        @Override
        public void onTimer(long timestamp, KeyedProcessFunction.OnTimerContext ctx, Collector out) throws Exception {
            ArrayList UrlViewCountArrayList = new ArrayList<>();
 
            for(UrlViewCount urlViewCount:urlViewCountListState.get()){
                UrlViewCountArrayList.add(urlViewCount);
            }
 
            UrlViewCountArrayList.sort(new Comparator() {
                @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(ctx.getCurrentKey()) +"\n");
 
            //取list前两个，包装信息输出
            for(int i=0;i<2;i++){
                UrlViewCount currTuple = UrlViewCountArrayList.get(i);
                String info="NO."+(i+1)+" "+"url: "+currTuple.url+" "+"访问量："+currTuple.count+" \n";
                result.append(info);
            }
            result.append("-------------------------\n");
 
            out.collect(result.toString());
        }
    }
 
}
```
相关阅读:
元宇宙电商-NFG系统，让你的数字藏品得到保障
 【Cocos2D -x | C++】CCNODe学习 |几个常见报错解决 |cocos2d-x“无法打开源文件”
ES:模板字符串的使用
 文件的导出
 Linux Mint 21.2 “EDGE” ISO 发布
 景联文科技：打造亿级高质量教育题库，赋能教育大语言模型新未来
 Shifu+WasmEdge：物联网数据轻松“瘦身”
《下一代互联网（IPv6）搭建与运维》
单例模式深入
 【C语言百炼成神】功法四 · 数组
原文地址：https://blog.csdn.net/JiaXingNashishua/article/details/126883698

7.1 基本处理函数（ProcessFunction）

7.1.1 处理函数的功能和使用

7.1.2 ProcessFunction 解析

7.1.3 处理函数的分类

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

7.2.1 定时器（Timer）和定时服务（TimerService）

7.2.2 KeyedProcessFunction 的使用

7.3 窗口处理函数

7.3.1 窗口处理函数的使用

7.3.2 ProcessWindowFunction 解析

7.4 应用案例——Top N

7.4.1 使用 ProcessAllWindowFunction

7.4.2 使用 KeyedProcessFunction