Flink处理函数 完整使用 (第七章)

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

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

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

本章主要讲解Flink中处理函数的使用方法。

一、基本处理函数（ProcessFunction）

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

1、处理函数的功能和使用

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

但是无论那种算子，如果我们想要访问事件的时间戳，或者当前的水位线信息，都是完全做不到的。在定义生成规则之后，水位线会源源不断地产生，像数据一样在任务间流动，可我们却不能像数据一样去处理它；跟时间相关的操作，目前我们只会用窗口来处理。而在很多应用需求中，要求我们对时间有更精细的控制，需要能够获取水位线，甚至要“把控时间”、定义什么时候做什么事，这就不是基本的时间窗口能够实现的了。

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

处理函数的使用与基本的转换操作类似，只需要直接基于DataStream调用.process()方法就可以了。方法需要传入一个ProcessFunction作为参数，用来定义处理逻辑。

这里ProcessFunction不是接口，而是一个抽象类，继承了AbstractRichFunction；MyProcessFunction是它的一个具体实现。所以所有的处理函数，都是富函数（RichFunction），富函数可以调用的东西这里同样都可以调用。

1) 编码

stream.process(new MyProcessFunction())
1

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class ProcessFunctionExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                )
                .process(new ProcessFunction<Event, String>() {
                    @Override
                    public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
                        if (value.user.equals("Mary")) {
                            out.collect(value.user);
                        } else if (value.user.equals("Bob")) {
                            out.collect(value.user);
                            out.collect(value.user);
                        }
                        System.out.println(ctx.timerService().currentWatermark());
                    }
                })
                .print();

        env.execute();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

这里我们在ProcessFunction中重写了.processElement()方法，自定义了一种处理逻辑：当数据的user为“Mary”时，将其输出一次；而如果为“Bob”时，将user输出两次。这里的输出，是通过调用out.collect()来实现的。另外我们还可以调用ctx.timerService().currentWatermark()来获取当前的水位线打印输出。所以可以看到，ProcessFunction函数有点像FlatMapFunction的升级版。可以实现Map、Filter、FlatMap的所有功能。很明显，处理函数非常强大，能够做很多之前做不到的事情。

接下来我们就深入ProcessFunction内部来进行详细了解。

2、ProcessFunction解析

在源码中我们可以看到，抽象类ProcessFunction继承了AbstractRichFunction，有两个泛型类型参数：I表示Input，也就是输入的数据类型；O表示Output，也就是处理完成之后输出的数据类型。
内部单独定义了两个方法：一个是必须要实现的抽象方法.processElement()；另一个是非抽象方法.onTimer()。

1）源码

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
2
3
4
5
6

2）抽象方法.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);
}
1
2
3
4
5

out：“收集器”（类型为Collector），用于返回输出数据。使用方式与flatMap算子中的收集器完全一样，直接调用out.collect()方法就可以向下游发出一个数据。这个方法可以多次调用，也可以不调用。

通过几个参数的分析不难发现，ProcessFunction可以轻松实现flatMap这样的基本转换功能（当然map、filter更不在话下）；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了

3）非抽象方法.onTimer()

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

与.processElement()类似，定时方法.onTimer()也有三个参数：时间戳（timestamp），上下文（ctx），以及收集器（out）。这里的timestamp是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService），以及任意输出处理之后的数据。

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

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

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

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的具体用法展开详细说明。

二、按键分区处理函数（KeyedProcessFunction）

在Flink程序中，为了实现数据的聚合统计，或者开窗计算之类的功能，我们一般都要先用keyBy算子对数据流进行“按键分区”，得到一个KeyedStream。也就是指定一个键（key），按照它的哈希值（hash code）将数据分成不同的“组”，然后分配到不同的并行子任务上执行计算；这相当于做了一个逻辑分流的操作，从而可以充分利用并行计算的优势实时处理海量数据。
只有在KeyedStream中才支持使用TimerService设置定时器的操作。所以一般情况下，我们都是先做了keyBy分区之后，再去定义处理操作；代码中更加常见的处理函数是KeyedProcessFunction，最基本的ProcessFunction反而出镜率没那么高。
接下来我们就先从定时服务（TimerService）入手，详细讲解KeyedProcessFunction的用法

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

KeyedProcessFunction的一个特色，就是可以灵活地使用定时器。
定时器（timers）是处理函数中进行时间相关操作的主要机制。 在.onTimer()方法中可以实现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能，是通过上下文中提供的“定时服务”（TimerService）来实现的。
定时服务与当前运行的环境有关。前面已经介绍过，ProcessFunction的上下文（Context）中提供了.timerService()方法，可以直接返回一个TimerService对象：

public abstract TimerService timerService();
1

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);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

六个方法可以分成两大类：基于处理时间和基于事件时间。而对应的操作主要有三个：获取当前时间，注册定时器，以及删除定时器。需要注意，尽管处理函数中都可以直接访问TimerService，不过只有基于KeyedStream的处理函数，才能去调用注册和删除定时器的方法；未作按键分区的DataStream不支持定时器操作，只能获取当前时间。

对于处理时间和事件时间这两种类型的定时器，TimerService内部会用一个优先队列将它们的时间戳（timestamp）保存起来，排队等待执行。可以认为，定时器其实是KeyedStream上处理算子的一个状态，它以时间戳作为区分。所以TimerService会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个key和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。这样一来，我们在代码中就方便了很多，可以肆无忌惮地对一个key注册定时器，而不用担心重复定义——因为一个时间戳上的定时器只会触发一次。

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

利用这个特性，有时我们可以故意降低时间戳的精度，来减少定时器的数量，从而提高处理性能。比如我们可以在设置定时器时只保留整秒数，那么定时器的触发频率就是最多1秒一次。

long coalescedTime = time / 1000 * 1000;
ctx.timerService().registerProcessingTimeTimer(coalescedTime);
1
2

这里注意定时器的时间戳必须是毫秒数，所以我们得到整秒之后还要乘以1000。定时器默认的区分精度是毫秒。

另外Flink对.onTimer()和.processElement()方法是同步调用的（synchronous），所以也不会出现状态的并发修改。

Flink的定时器同样具有容错性，它和状态一起都会被保存到一致性检查点（checkpoint）中。当发生故障时，Flink会重启并读取检查点中的状态，恢复定时器。如果是处理时间的定时器，有可能会出现已经“过期”的情况，这时它们会在重启时被立刻触发。关于Flink的检查点和容错机制

1）KeyedProcessFunction的使用

KeyedProcessFunction可以说是处理函数中的“嫡系部队”，可以认为是ProcessFunction的一个扩展。我们只要基于keyBy之后的KeyedStream，直接调用.process()方法，这时需要传入的参数就是KeyedProcessFunction的实现类。

1、案例1（处理时间）

package com.example.chapter07;

import com.example.chapter05.ClickSource;
import com.example.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.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;


/**
 * 处理时间
 * 

 * 确定不用事件时间的话 WatermarkStrategy 没用了
 */
public class ProcessingTimeTimerTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new CustomSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                );

        //以user为键 分组
        /**
         * new KeyedProcessFunction
         *     String key 的类型
         *     Event 输入
         *     String输出
         */
        stream.keyBy(data -> data.user).process(new KeyedProcessFunction<String, Event, String>() {

            @Override
            public void processElement(Event value, Context ctx, Collector<String> out) {

                Long currTs = ctx.timestamp();
                out.collect(ctx.getCurrentKey() + "数据到达，时间戳： " + new Timestamp(currTs) +"watermark " + ctx.timerService().currentWatermark());

                ctx.timerService().registerEventTimeTimer(currTs + 10 * 1000L);
            }


            /**
             * 定义的定时器 什么时间响
             * onTimer 没有数据、靠时间来触发的
             * 可以获取到当前定这个定时器、定他的那个时候的 那个的时间
             *
             *
             * @param timestamp
             * @param ctx
             * @param out
             * @throws Exception
             */
            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {

                out.collect(ctx.getCurrentKey() + " 定时器触发, 触发时间： " + new Timestamp(timestamp) + "watermark " + ctx.timerService().currentWatermark());
            }
        }).print();

        env.execute();

    }

    //自定义测试数据
    public static class CustomSource implements SourceFunction<Event> {

        @Override
        public void run(SourceContext<Event> ctx) throws Exception {
            //直接发出测试数据
            ctx.collect(new Event("Mary", "./home", 1000L));
            Thread.sleep(5000L);
            ctx.collect(new Event("Alice", "./home", 11000L));
            Thread.sleep(5000L);

            ctx.collect(new Event("Bob", "./home", 11001L));
            Thread.sleep(5000L);
        }

        @Override
        public void cancel() {

        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

2）案例2（事件定时器）

package com.example.chapter07;

import com.example.chapter05.ClickSource;
import com.example.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.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;
import java.time.Duration;

/**
 * 事件定时器
 * KeyedProcessFunction
 */
public class EventTimeTimerTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                );


        //以user为键 分组
        /**
         * new KeyedProcessFunction
         *     String key 的类型
         *     Event 输入
         *     String输出
         */
        stream.keyBy(data -> data.user)
                .process(new KeyedProcessFunction<String, Event, String>() {

                    @Override
                    public void processElement(Event value, Context ctx, Collector<String> out) {
                        long currTs = ctx.timestamp(); //数据自带的时间戳
                        out.collect(ctx.getCurrentKey() + "数据到达，时间戳： " + new Timestamp(currTs) +
                                "watermark: " + ctx.timerService().currentWatermark());

                        //注册一个10秒后的定时器
                        ctx.timerService().registerEventTimeTimer(currTs + 10 * 1000L);
                    }


                    /**
                     * 定义的定时器 什么时间响
                     * onTimer 没有数据、靠时间来触发的
                     * 可以获取到当前定这个定时器、定他的那个时候的 那个的时间
                     * @param timestamp
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
                        out.collect(ctx.getCurrentKey() + " 定时器触发, 触发时间： " +
                                new Timestamp(timestamp) +
                                "watermark: " + ctx.timerService().currentWatermark()
                        );
                    }
                }).print();

        env.execute();

    }

    //自定义测试数据源
    public static class CustomSource implements SourceFunction<Event> {

        @Override
        public void run(SourceContext<Event> ctx) throws Exception {
            ctx.collect(new Event("Mary", "./home", 1000L));
            Thread.sleep(5000L);
            ctx.collect(new Event("Alice", "./home", 1100L));
        }

        @Override
        public void cancel() {

        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

Mary数据到达，时间戳： 1970-01-01 08:00:01.0watermark -9223372036854775808
Alice数据到达，时间戳： 1970-01-01 08:00:11.0watermark 999
Bob数据到达，时间戳： 1970-01-01 08:00:11.001watermark 10999
Mary 定时器触发, 触发时间： 1970-01-01 08:00:11.0watermark 11000
Alice 定时器触发, 触发时间： 1970-01-01 08:00:21.0watermark 9223372036854775807
Bob 定时器触发, 触发时间： 1970-01-01 08:00:21.001watermark 9223372036854775807
1
2
3
4
5
6

每来一条数据，都会输出两行“数据到达”的信息，并以分割线隔开；两条数据到达的时间间隔为5秒。当第三条数据到达后，随后立即输出一条定时器触发的信息；再过5秒之后，剩余两条定时器信息输出，程序运行结束。
我们可以发现，数据到来之后，当前的水位线与时间戳并不是一致的。当第一条数据到来，时间戳为1000，可水位线的生成是周期性的（默认200ms一次），不会立即发生改变，所以依然是最小值Long.MIN_VALUE；随后只要到了水位线生成的时间点（200ms到了），就会依据当前的最大时间戳1000来生成水位线了。这里我们没有设置水位线延迟，默认需要减去1毫秒，所以水位线推进到了999。而当时间戳为11000的第二条数据到来之后，水位线同样没有立即改变，仍然是999，就好像总是“滞后”数据一样。

这样程序的行为就可以得到合理解释了。事件时间语义下，定时器触发的条件就是水位线推进到设定的时间。第一条数据到来后，设定的定时器时间为1000 + 10 * 1000 = 11000；而当时间戳为11000的第二条数据到来，水位线还处在999的位置，当然不会立即触发定时器；而之后水位线会推进到10999，同样是无法触发定时器的。必须等到第三条数据到来，将水位线真正推进到11000，就可以触发第一个定时器了。第三条数据发出后再过5秒，没有更多的数据生成了，整个程序运行结束将要退出，此时Flink会自动将水位线推进到长整型的最大值（Long.MAX_VALUE）。于是所有尚未触发的定时器这时就统一触发了，我们就在控制台看到了后两个定时器的触发信息。

三、窗口处理函数

除了KeyedProcessFunction，另外一大类常用的处理函数，就是基于窗口的ProcessWindowFunction和ProcessAllWindowFunction了。会发现我们之前已经简单地使用过窗口处理函数了。

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())
1
2
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 {...}
}
1
2
3
4
5
6
7
8

ProcessWindowFunction依然是一个继承了AbstractRichFunction的抽象类，它有四个类型参数：

IN：input，数据流中窗口任务的输入数据类型。
OUT：output，窗口任务进行计算之后的输出数据类型。
KEY：数据中键key的类型。
W：窗口的类型，是Window的子类型。一般情况下我们定义时间窗口，W就是TimeWindow。
1
2
3
4

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

key：窗口做统计计算基于的键，也就是之前keyBy用来分区的字段。
context：当前窗口进行计算的上下文，它的类型就是ProcessWindowFunction内部定义的抽象类Context。
elements：窗口收集到用来计算的所有数据，这是一个可迭代的集合类型。
out：用来发送数据输出计算结果的收集器，类型为Collector。
1
2
3
4

可以明显看出，这里的参数不再是一个输入数据，而是窗口中所有数据的集合。而上下文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);
}
1
2
3
4
5
6
7
8
9
10

除了可以通过.output()方法定义侧输出流不变外，其他部分都有所变化。这里不再持有TimerService对象，只能通过currentProcessingTime()和currentWatermark()来获取当前时间，所以失去了设置定时器的功能； 另外由于当前不是只处理一个数据，所以也不再提供.timestamp()方法。与此同时，也增加了一些获取其他信息的方法：比如可以通过.window()直接获取到当前的窗口对象，也可以通过.windowState()和.globalState()获取到当前自定义的窗口状态和全局状态。注意这里的“窗口状态”是自定义的，不包括窗口本身已经有的状态，针对当前key、当前窗口有效；而“全局状态”同样是自定义的状态，针对当前key的所有窗口有效。

所以我们会发现，ProcessWindowFunction中除了.process()方法外，并没有.onTimer()方法，而是多出了一个.clear()方法。从名字就可以看出，这主要是方便我们进行窗口的清理工作。如果我们自定义了窗口状态，那么必须在.clear()方法中进行显式地清除，避免内存溢出。

这里有一个问题：没有了定时器，那窗口处理函数就失去了一个最给力的武器，如果我们希望有一些定时操作又该怎么做呢？其实仔细思考会发现，对于窗口而言，它本身的定义就包含了一个触发计算的时间点，其实一般情况下是没有必要再去做定时操作的。如果非要这么干，Flink也提供了另外的途径——使用窗口触发器（Trigger）。在触发器中也有一个TriggerContext，它可以起到类似TimerService的作用：获取当前时间、注册和删除定时器，另外还可以获取当前的状态。这样设计无疑会让处理流程更加清晰——定时操作也是一种“触发”，所以我们就让所有的触发操作归触发器管，而所有处理数据的操作则归窗口函数管。

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

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

1）编码 ProcessWindowsFunction

package com.example.chapter06;


import com.example.chapter05.ClickSource;
import com.example.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.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
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.HashSet;

/**
 * ProcessWindowsFunction 主要用这个个
 * 处理窗口函数（ProcessWindowFunction）
 */
public class WindowProcessTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(100); //去设置生成水位线的时间间隔

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                //TODO 乱序流的watermark生成 ----forBoundedOutOfOrderness
                //插入水位线
                .assignTimestampsAndWatermarks(WatermarkStrategy.
                        //针对乱序流插入水位线，延迟时间设置5
                                <Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {

                            //抽取时间戳的逻辑
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        })
                );

        stream.print("input");

        stream.keyBy(data -> true)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new UvCountByWindows()).print();

        env.execute();
    }


    //实现自定义的ProcessWindowFunction()
    //        INPUT-> Event
    //        OUTPUT-> String
    //        KEY-> String
    public static class UvCountByWindows extends ProcessWindowFunction<Event, String, Boolean, TimeWindow> {


        /**
         * Collector out 输出
         * Iterable elements 输入
         * @param aBoolean
         * @param context
         * @param elements
         * @param out
         * @throws Exception
         */
        @Override
        public void process(Boolean aBoolean, ProcessWindowFunction<Event, String, Boolean, TimeWindow>.Context context, Iterable<Event> elements, Collector<String> out) throws Exception {
            //用一个HashSet 保存user
            HashSet<String> userSet = new HashSet<>();

            //从element中遍历数据，放到Set中去重
            for (Event event : elements) {
                userSet.add(event.user);
            }


            Integer uv = userSet.size();

            //结合窗口信息
            long start = context.window().getStart();
            long end = context.window().getEnd();

            out.collect("窗口 " + new Timestamp(start) + " ~ " + new Timestamp(end)
                    + "UV值为：" + uv
            );

        }
    }

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

四、应用案例——Top N

1、使用ProcessAllWindowFunction

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

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

编码 使用ProcessAllWindowFunction

package com.example.chapter07;

import com.example.chapter05.ClickSource;
import com.example.chapter05.Event;
import com.example.chapter06.UrlCountViewExample;
import com.example.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.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;

/**
 * 使用ProcessAllWindowFunction
 * 的基本思路
 */
public class TopNExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        })
                );


        //1、按照URL分组，统计每个窗口内每个URL的访问量
        //aggregate 增量聚合函数
        SingleOutputStreamOperator<UrlViewCount> 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 TopProcessResult(2))
                .print();


        env.execute();

    }

    public static class TopProcessResult extends KeyedProcessFunction<Long, UrlViewCount, String> {

        // 定义一个属性 n
        private Integer n;

        //定义列表状态
        private ListState<UrlViewCount> urlViewCountListState;


        public TopProcessResult(Integer n) {
            this.n = n;
        }

        //在环境中获取状态


        @Override
        public void open(Configuration parameters) throws Exception {
            urlViewCountListState = getRuntimeContext().
                    getListState(new ListStateDescriptor<UrlViewCount>(
                            "url-count-list", Types.POJO(UrlViewCount.class)));
        }

        @Override
        public void processElement(UrlViewCount value, KeyedProcessFunction<Long, UrlViewCount, String>.Context ctx, Collector<String> collector) throws Exception {
            // 将数据保存到状态中
            urlViewCountListState.add(value);
            //注册windowsEnd +1 ms的定时器
            ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<Long, UrlViewCount, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            ArrayList<UrlViewCount> urlViewCountArrayList = new ArrayList<>();

            for (UrlViewCount viewCount : urlViewCountListState.get()) {
                urlViewCountArrayList.add(viewCount);
            }
            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("-------------------");
            result.append("窗口结束时间：" + new Timestamp(ctx.getCurrentKey()) + "\n");

            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());

        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

2）编码 ProcessAllWindowFunction结合AggregateFunction

方式二

package com.example.chapter07;

import com.example.chapter05.ClickSource;
import com.example.chapter05.Event;
import com.example.chapter06.UrlCountViewExample;
import com.example.chapter06.UrlViewCount;
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;

/**
 * 使用ProcessAllWindowFunction 二
 * 的基本思路 代码测试和实现
 *
 * 把全部数据放到了 windowAll
 */
public class TopNExample_ProcessAllWindowFunction {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @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 UrlAllWindowsResult())
                .print();


        env.execute();
    }


    //实现自定义全窗口函数，包装信息输出结果
    public static class UrlAllWindowsResult extends ProcessAllWindowFunction<ArrayList<Tuple2<String, Long>>, String, TimeWindow> {


        @Override
        public void process(ProcessAllWindowFunction<ArrayList<Tuple2<String, Long>>, String, TimeWindow>.Context context, Iterable<ArrayList<Tuple2<String, Long>>> element, Collector<String> collector) throws Exception {
            ArrayList<Tuple2<String, Long>> next = element.iterator().next();
            StringBuilder result = new StringBuilder();
            result.append("-------------------");
            result.append("窗口结束时间：" + new Timestamp(context.window().getEnd()) + "\n");
            //取list前两个，包装信息输出
            for (int i = 0; i < 2; i++) {
                Tuple2<String, Long> currTuple = next.get(i);
                String info = "No. " + (i + 1) + " "
                        +"url: " +currTuple.f0+" "
                        +"访问量：" +currTuple.f1 +"\n";

                result.append(info);
            }
            result.append("_______________________\n");
            collector.collect(result.toString());
        }
    }


    //实现自定义的增量聚合函数

    public static class UrlHashMapCountAgg implements AggregateFunction<String, HashMap<String, Long>, ArrayList<Tuple2<String, Long>>> {

        @Override
        public HashMap<String, Long> createAccumulator() {
            return new HashMap<>();
        }

        @Override
        public HashMap<String, Long> add(String value, HashMap<String, Long> accumulator) {
            //如果包含这个value +1
            if (accumulator.containsKey(value)) {
                Long count = accumulator.get(value);
                accumulator.put(value, count + 1);
            } else {
                accumulator.put(value, 1L);
            }
            return accumulator;
        }

        @Override
        public ArrayList<Tuple2<String, Long>> getResult(HashMap<String, Long> accumulator) {
            ArrayList<Tuple2<String, Long>> list = new ArrayList();

            //获取key,值
            for (String key : accumulator.keySet()) {
                list.add(Tuple2.of(key, accumulator.get(key)));
            }

            //排序
            list.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();
                }
            });
            return list;
        }

        @Override
        public HashMap<String, Long> merge(HashMap<String, Long> accumulator, HashMap<String, Long> acc1) {
            return null;
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132

3）侧输出流（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));
      }
});
1
2
3
4
5
6
7
8
9
10

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

我们可以在外部先将OutputTag声明出来：
OutputTag outputTag = new OutputTag(“side-output”) {};

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

五、本章总结

Flink拥有非常丰富的多层API，而底层的处理函数可以说是最为强大、最为灵活的一种。广义上来讲，处理函数也可以认为是DataStream API中的一部分，它的调用方式与其他转换算子完全一致。处理函数可以访问时间、状态，定义定时操作，它可以直接触及流处理最为本质的组成部分。所以处理函数不仅是我们处理复杂需求时兜底的“大招”，更是理解流处理本质的重要一环。

在本章中，我们详细介绍了处理函数的功能和底层的结构，重点讲解了最为常用的KeyedProcessFunction和ProcessWindowFunction，并实现了电商应用中Top N的经典案例，另外还介绍了侧输出流的用法。而关于合并两条流之后的处理函数，以及广播连接流（BroadcastConnectedStream）的处理操作，调用方法和原理都非常类似，我们会在后续章节继续展开。