杰霸霸博客

主题

处理函数 ​

ᅟᅠᅟᅠ之前所介绍的流处理API，无论是基本的转换、聚合，还是更为复杂的窗口操作，其实都是基于DataStream进行转换的，所以可以统称为DataStream API。在Flink更底层，我们可以不定义任何具体的算子(比如map，filter，或者window)，而只是提炼出一个统一的"处理"(process)操作——它是所有转换算子的一个概括性的表达，可以自定义处理逻辑，所以这一层接口就被叫作"处理函数"(process function)。
Alt text

1. 基本处理函数(ProcessFunction) ​

1.1 处理函数的功能和使用 ​

ᅟᅠᅟᅠ我们之前学习的转换算子，一般只是针对某种具体操作来定义的，能够拿到的信息比较有限。如果我们想要访问事件的时间戳，或者当前的水位线信息，都是完全做不到的。跟时间相关的操作，目前我们只会用窗口来处理。而在很多应用需求中，要求我们对时间有更精细的控制，需要能够获取水位线，甚至要"把控时间"、定义什么时候做什么事，这就不是基本的时间窗口能够实现的了。

ᅟᅠᅟᅠ这时就需要使用底层的处理函数。处理函数提供了一个"定时服务"(TimerService)，我们可以通过它访问流中的事件(event)、时间戳(timestamp)、水位线(watermark)，甚至可以注册"定时事件"。而且处理函数继承了AbstractRichFunction抽象类，所以拥有富函数类的所有特性，同样可以访问状态(state)和其他运行时信息。此外，处理函数还可以直接将数据输出到侧输出流(side output)中。所以，处理函数是最为灵活的处理方法，可以实现各种自定义的业务逻辑。

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

1.2 ProcessFunction解析 ​

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

java
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()。
  • out："收集器"(类型为Collector)，用于返回输出数据。使用方式与flatMap算子中的收集器完全一样，直接调用out.collect()方法就可以向下游发出一个数据。这个方法可以多次调用，也可以不调用。
    通过几个参数的分析不难发现，ProcessFunction可以轻松实现flatMap、map、filter这样的基本转换功能；而通过富函数提供的获取上下文方法getRuntimeContext()，也可以自定义状态(state)进行处理，这也就能实现聚合操作的功能了。
  1. 非抽象方法onTimer()
    这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过"定时服务"TimerService来注册的。打个比方，注册定时器(timer)就是设了一个闹钟，到了设定时间就会响；而onTimer()中定义的，就是闹钟响的时候要做的事。所以它本质上是一个基于时间的"回调"(callback)方法，通过时间的进展来触发；在事件时间语义下就是由水位线(watermark)来触发了。定时方法onTimer()也有三个参数：时间戳(timestamp)，上下文(ctx)，以及收集器(out)。这里的timestamp是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务(TimerService)，以及任意输出处理之后的数据。既然有onTimer()方法做定时触发，我们用ProcessFunction也可以自定义数据按照时间分组、定时触发计算输出结果；这其实就实现了窗口(window)的功能。所以说ProcessFunction其实可以实现一切功能。

警告

在Flink中，只有"按键分区流"KeyedStream才支持设置定时器的操作。

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

4. 按键分区处理函数(KeyedProcessFunction) ​

在上节中提到，只有在KeyedStream中才支持使用TimerService设置定时器的操作。所以一般情况下，我们都是先做了keyBy分区之后，再去定义处理操作；代码中更加常见的处理函数是KeyedProcessFunction。

4.1 定时器(Timer)和定时服务(TimerService) ​

在onTimer()方法中可以实现定时处理的逻辑，而它能触发的前提，就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能，是通过上下文中提供的"定时服务"来实现的。定时服务与当前运行的环境有关。前面已经介绍过，ProcessFunction的上下文(Context)中提供了timerService()方法，可以直接返回一个TimerService对象。TimerService是Flink关于时间和定时器的基础服务接口，包含以下六个方法：

java
// 获取当前的处理时间
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会以键(key)和时间戳为标准，对定时器进行去重；也就是说对于每个key和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。

4.2 KeyedProcessFunction案例 ​

基于keyBy之后的KeyedStream，直接调用process()方法，这时需要传入的参数就是KeyedProcessFunction的实现类。
类似地，KeyedProcessFunction也是继承自AbstractRichFunction的一个抽象类，与ProcessFunction的定义几乎完全一样，区别只是在于类型参数多了一个K，这是当前按键分区的key的类型。同样地，我们必须实现一个processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法onTimer()，用来定义定时器触发时的回调操作。

  1. 事件时间处理案例：
java
public class FlinkKeyedProcessTimerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 设置并行度为 1
        env.setParallelism(1);

        DataStream<WaterSensor> s1 = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {

                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((value, ts)->value.getTs()*1000L));

        //1. 分别做Keyby，key其实就是关联条件
        KeyedStream<WaterSensor, String> ks1 = s1.keyBy(r -> r.getId());
        ks1.process(new KeyedProcessFunction<String, WaterSensor, String>() {

            /**
             * 来一条数据处理一次
             * @param value The input value.
             * @param ctx
             * @param out The collector for returning result values.
             * @throws Exception
             */
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                // 提取数据中的事件时间,没有就是null
                Long ts = ctx.timestamp();
                // 定时器
                TimerService timerService = ctx.timerService();
                String currentKey = ctx.getCurrentKey();
                // 定时器有两种：基于处理时间，基于事件时间，如果在5s内有多条数据，也就是注册定时器有多个，定时器只会执行一次(自动去重)
                // 注册定时器： 基于处理时间
                // timerService.registerProcessingTimeTimer();
                // 注册定时器： 基于事件时间
                timerService.registerEventTimeTimer(5000L);
                // 当前时间, 就是系统时间
                long currentProcessingTime = timerService.currentProcessingTime();
                // 删除定时器
                // timerService.deleteProcessingTimeTimer();
                // timerService.deleteEventTimeTimer();
                long currentWatermark = timerService.currentWatermark();
                System.out.println("当前时间是"+currentProcessingTime+", 注册了一个5s的定时器");
            }

            /**
             * 事件进展到定时器注册的时间，调用该方法
             * @param timestamp 当前时间进展
             * @param ctx 上下文
             * @param out 采集器
             * @throws Exception
             */
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                System.out.println("现在时间是"+timestamp+"定时器触发");
            }
        });
        env.execute();
    }
}

运行结果：
Alt text 2. 处理时间案例：

java
public class FlinkKeyedProcessTimerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        // 设置并行度为 1
        env.setParallelism(1);

        DataStream<WaterSensor> s1 = env.socketTextStream("hadoop102", 7777)
                .map(new MapFunction<String, WaterSensor>() {

                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((value, ts)->value.getTs()*1000L));

        //1. 分别做Keyby，key其实就是关联条件
        KeyedStream<WaterSensor, String> ks1 = s1.keyBy(r -> r.getId());
        ks1.process(new KeyedProcessFunction<String, WaterSensor, String>() {

            /**
             * 来一条数据处理一次
             * @param value The input value.
             * @param ctx
             * @param out The collector for returning result values.
             * @throws Exception
             */
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                // 提取数据中的事件时间
                Long ts = ctx.timestamp();
                // 定时器
                TimerService timerService = ctx.timerService();
                String currentKey = ctx.getCurrentKey();
                // 当前时间, 就是系统时间
                long currentTs = timerService.currentProcessingTime();
                long currentWatermark = timerService.currentWatermark();

                // 注册定时器： 基于处理时间
                // timerService.registerEventTimeTimer(5000L);
                // System.out.println("当前key="+currentKey+",当前时间是"+currentTs+", 注册了一个5s的定时器");
                // 注册定时器： 基于事件时间
                 timerService.registerProcessingTimeTimer(currentTs + 5000L);
                 System.out.println("当前key="+currentKey+",当前时间是"+currentTs+", 注册了一个5s后的定时器");
                // 删除定时器
                // timerService.deleteProcessingTimeTimer();
                // timerService.deleteEventTimeTimer();

            }

            /**
             * 事件进展到定时器注册的时间，调用该方法
             * @param timestamp 当前时间进展
             * @param ctx 上下文
             * @param out 采集器
             * @throws Exception
             */
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                System.out.println("现在的key="+ctx.getCurrentKey()+",现在时间是"+timestamp+"定时器触发");
            }
        });

        env.execute();
    }
}

运行结果：
Alt text

定时器总结

  1. keyBy才有
  2. 事件时间定时器，通过watermark来触发的
    watermark >= 注册时间
    注意： watermark = 当前最大事件时间-等待时间-1ms, 因为1ms,导致推迟一条数据。比如5s的定时器，等待3s, watermark = 8s -5s -1ms = 4999ms, 不会触发5s的定时器
  3. 在processElement()中可以获取当前watermark，可以看到当前水位线进展进行佐证：
java
@Override
public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
    // 提取数据中的事件时间
    Long ts = ctx.timestamp();
    // 定时器
    TimerService timerService = ctx.timerService();
    String currentKey = ctx.getCurrentKey();
    // 当前时间, 就是系统时间
    long currentTs = timerService.currentProcessingTime();
    long currentWatermark = timerService.currentWatermark();
    System.out.println("当前数据="+value+",当前水位线是"+currentWatermark);
    // 不用注册定时器，单纯看水位线进展情况
}

运行结果：
Alt text

5. 窗口处理函数 ​

除了KeyedProcessFunction，另外一大类常用的处理函数，就是基于窗口的ProcessWindowFunction和ProcessAllWindowFunction了。在前面窗口章节的介绍中，我们之前已经简单地使用过窗口处理函数了。

5.1 窗口处理函数的使用 ​

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

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

java
stream.keyBy( t -> t.f0 )
        .window( TumblingEventTimeWindows.of(Time.seconds(10)) )
        .process(new MyProcessWindowFunction())

5.2 ProcessWindowFunction解析 ​

ProcessWindowFunction既是处理函数又是全窗口函数。从名字上也可以推测出，它的本质似乎更倾向于"窗口函数"一些。事实上它的用法也确实跟其他处理函数有很大不同。我们可以从源码中的定义看到这一点：

java
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里面处理数据的核心方法process()。方法包含四个参数。
  • key：窗口做统计计算基于的键，也就是之前keyBy用来分区的字段。
  • context：当前窗口进行计算的上下文，它的类型就是ProcessWindowFunction内部定义的抽象类Context。
  • elements：窗口收集到用来计算的所有数据，这是一个可迭代的集合类型。
  • out：用来发送数据输出计算结果的收集器，类型为Collector。
    可以明显看出，这里的参数不再是一个输入数据，而是窗口中所有数据的集合。而上下文context所包含的内容也跟其他处理函数有所差别：
java
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);

除了可以通过output()方法定义侧输出流不变外，其他部分都有所变化。这里不再持有TimerService对象，只能通过currentProcessingTime()和currentWatermark()来获取当前时间，所以失去了设置定时器的功能；另外由于当前不是只处理一个数据，所以也不再提供timestamp()方法。与此同时，也增加了一些获取其他信息的方法：比如可以通过window()直接获取到当前的窗口对象，也可以通过windowState()和globalState()获取到当前自定义的窗口状态和全局状态。注意这里的"窗口状态"是自定义的，不包括窗口本身已经有的状态，针对当前key、当前窗口有效；而"全局状态"同样是自定义的状态，针对当前key的所有窗口有效。
所以我们会发现，ProcessWindowFunction中除了process()方法外，并没有onTimer()方法，而是多出了一个clear()方法。从名字就可以看出，这主要是方便我们进行窗口的清理工作。如果我们自定义了窗口状态，那么必须在.clear()方法中进行显式地清除，避免内存溢出。

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

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

6. 应用案例——Top N ​

需求：实时统计一段时间内的出现次数最多的水位。例如，统计最近10秒钟内出现次数最多的两个水位，并且每5秒钟更新一次。我们知道，这可以用一个滑动窗口来实现。于是就需要开滑动窗口收集传感器的数据，按照不同的水位进行统计，而后汇总排序并最终输出前两名。这其实就是著名的"Top N"问题。

6.1 实现方式一 ​

使用ProcessAllWindowFunction去实现，不区分不同水位，而是将所有访问数据都收集起来，统一进行统计计算。所以可以不做keyBy，直接基于DataStream开窗，然后使用全窗口函数ProcessAllWindowFunction来进行处理。

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

java
public class TopNDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 7777)
                .map((MapFunction<String, WaterSensor>) value -> {
                    String[] split = value.split(",");
                    return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                })
                // 处理最多延迟3秒的乱序事件
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((value, ts) -> value.getTs() * 1000L));
        // 最近10s --窗口长度  每5s -- 滑动步长
        sensorDS.windowAll(SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofSeconds(5)))
                .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
                    @Override
                    public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        LinkedHashMap<Integer, Integer> hashMap = new LinkedHashMap<>();
                        // 1. 遍历数据 统计各个vc出现的累计次数
                        for (WaterSensor element : elements) {
                            Integer elementVc = element.getVc();
                            // 1.1 如果不是第一条数据
                            if (hashMap.containsKey(elementVc)) {
                                Integer count = hashMap.get(elementVc);
                                hashMap.put(elementVc, count + 1);
                            } else {
                                // 1.2 如果是第一条数据
                                hashMap.put(elementVc, 1);
                            }
                        }
                        ArrayList<Tuple2<Integer, Integer>> list = new ArrayList<>();
                        hashMap.forEach((k, v) -> {
                            list.add(new Tuple2<>(k, v));
                        });
                        // 降序
                        list.sort((o1, o2) -> o2.f1 - o1.f1);
                        // 取前2个
                        StringBuffer sb = new StringBuffer(8);
                        sb.append("=========Top2===========\n");
                        for (int i = 0; i < Math.min(list.size(), 2); i++) {
                            Tuple2<Integer, Integer> tuple2 = list.get(i);
                            sb.append("vc=" + tuple2.f0 + ",count=" + tuple2.f1);
                            sb.append(",窗口结束时间=" + DateFormatUtils.format(context.window().getEnd(), "yyyy-MM-dd HH:mm:ss") + "\n");
                        }
                        sb.append("========================");
                        out.collect(sb.toString());
                    }
                }).print();
        env.execute();
    }
}

输入测试：
Alt text

6.1 实现方式二 ​

使用KeyedProcessFunction，在上面的实现过程中，我们没有进行按键分区，直接将所有数据放在一个分区上进行了开窗操作。这相当于将并行度强行设置为1，在实际应用中是要尽量避免的，所以Flink官方也并不推荐使用AllWindowedStream进行处理。另外，我们在全窗口函数中定义了HashMap来统计vc的出现次数，计算过程是要先收集齐所有数据、然后再逐一遍历更新HashMap，这显然不够高效。

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

java
public class TopNDemo2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 7777)
                .map((MapFunction<String, WaterSensor>) value -> {
                    String[] split = value.split(",");
                    return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                })
                // 处理最多延迟3秒的乱序事件
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((value, ts) -> value.getTs() * 1000L));
        // 按照不同的vc去keyBy, 分别去count
        /**
         * 1. 按照vc做keyBy,开窗, 分别count
         *     ==> 增量聚合，计算count
         *     ==> 全窗口，对计算结果count值封装
         *          ==> 为了让同一个窗口时间范围的计算结果到一起去
         * 2. 对同一个窗口范围的count值进行处理： 排序、取前N个
         *      ==> 按照 windowEnd做keyBy
         *      ==> 使用process, 来一条调用一次，需要先存，分开存HashMap,key=windowEnd,value=list
         *          ==> 使用定时器，对存起来的结果进行排序、取前N个
         */
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> windowAggregate = sensorDS.keyBy(WaterSensor::getVc)
                // 最近10s --窗口长度  每5s -- 滑动步长
                .window(SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofSeconds(5)))
                // 开窗聚合后就是普通的流，没有了窗口信息
                .aggregate(new AggregateFunction<WaterSensor, Integer, Integer>() {
                    // 初始值
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    // 累加器  也就是累计方式
                    @Override
                    public Integer add(WaterSensor value, Integer accumulator) {
                        return accumulator + 1;
                    }

                    @Override
                    public Integer getResult(Integer accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return null;
                    }
                }, new ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer vc, ProcessWindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
                        Integer count = elements.iterator().next();
                        out.collect(new Tuple3<>(vc, count, context.window().getEnd()));
                    }
                });
        windowAggregate.keyBy(r -> r.f2)
                .process(new KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>() {
                    // 保存不同窗口的统计结果，key是窗口的结束时间, value是list数据
                    private LinkedHashMap<Long, ArrayList<Tuple3<Integer, Integer, Long>>> map = new LinkedHashMap<>();
                    // 要取topN的数量
                    private int topN = 2;

                    @Override
                    public void processElement(Tuple3<Integer, Integer, Long> value, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        // 1. 把数据存入到map中
                        Long windowEnd = value.f2;
                        if (map.containsKey(windowEnd)) {
                            // 1.1 如果已经存在vc，直接添加
                            ArrayList<Tuple3<Integer, Integer, Long>> list = map.get(windowEnd);
                            list.add(value);
                        } else {
                            // 1.2 如果不存在，创建一个新的list，添加进去
                            ArrayList<Tuple3<Integer, Integer, Long>> list = new ArrayList<>();
                            list.add(value);
                            map.put(windowEnd, list);
                        }
                        // 注册一个定时器, 异步延迟1ms
                        ctx.timerService().registerEventTimeTimer(windowEnd + 1);
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        // 2. 取出所有的list数据
                        Long currentKey = ctx.getCurrentKey();
                        ArrayList<Tuple3<Integer, Integer, Long>> list = map.get(currentKey);
                        list.sort((o1, o2) -> o2.f1 - o1.f1);
                        // 3. 取前topN个
                        StringBuilder sb = new StringBuilder();
                        sb.append("=========Top"+topN+"===========\n");
                        for (int i = 0; i < Math.min(list.size(), topN); i++) {
                            Tuple3<Integer, Integer, Long> tuple3 = list.get(i);
                            sb.append("vc="+tuple3.f0 + ", count=" + tuple3.f1);
                            sb.append(",窗口结束时间=" + DateFormatUtils.format(tuple3.f2, "yyyy-MM-dd HH:mm:ss") + "\n");
                        }
                        // 用完的list, 及时清理
                        list.clear();
                        out.collect(sb.toString());
                    }
                }).print();
        env.execute();
    }
}

运行效果：
Alt text

7. 侧输出流 ​

处理函数还有另外一个特有功能，就是将自定义的数据放入"侧输出流"(side output)输出。这个概念我们并不陌生，之前在讲到窗口处理迟到数据时，最后一招就是输出到侧输出流。而这种处理方式的本质，其实就是处理函数的侧输出流功能。

我们之前讲到的绝大多数转换算子，输出的都是单一流，流里的数据类型只能有一种。而侧输出流可以认为是"主流"上分叉出的"支流"，所以可以由一条流产生出多条流，而且这些流中的数据类型还可以不一样。利用这个功能可以很容易地实现"分流"操作。具体应用时，只要在处理函数的processElement()或者onTimer()方法中，调用上下文的output()方法就可以了。

java
DataStream<Integer> stream = env.fromSource(...);

OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

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

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

7.1 实现需求 ​

对每个传感器，水位超过10的输出告警信息。
实现代码如下：

java
public class SideOutputDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("hadoop102", 7777)
                .map((MapFunction<String, WaterSensor>) value -> {
                    String[] split = value.split(",");
                    return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                })
                // 处理最多延迟3秒的乱序事件
                .assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((value, ts) -> value.getTs() * 1000L));
        OutputTag<String> warnOutputTag = new OutputTag<>("warn", Types.STRING);
        SingleOutputStreamOperator<String> process = sensorDS.keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor input, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        if (input.getVc() < 10) {
                            out.collect(input.toString());
                        } else {
                            ctx.output(warnOutputTag, "当前vc大于阈值：" + input);
                        }
                    }
                });
        process.getSideOutput(warnOutputTag).printToErr("warn");
        process.print("主流");
        env.execute();
    }
}

运行效果：
Alt text