Flink处理函数

1、简介

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

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

2、基本处理函数(ProcessFunction)

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

2.1 处理函数的功能和使用

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

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

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

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

    1
    stream.process(new MyProcessFunction())
    • 这里ProcessFunction不是接口,而是一个抽象类,继承了AbstractRichFunction; MyProcessFunction 是它的一个具体实现。所以所有的处理函数,都是富函数(RichFunction),富函数可以调用的东西这里同样都可以调用。
    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
    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>forBoundedOutOfOrderness(Duration.ZERO)
    .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
    @Override
    public long extractTimestamp(Event element, long recordTimestamp) {
    return element.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();
    }
    }
    • 这里在ProcessFunction中重写了.processElement()方法,自定义了一种处理逻辑:当数据的user为“Mary”时,将其输出一次;而如果为“Bob”时,将user输出两次。这里的输 出,是通过调用out.collect()来实现的。另外我们还可以调用ctx.timerService().currentWatermark()来获取当前的水位线打印输出。所以可以看到,ProcessFunction函数有点像FlatMapFunction的升级版。可以实现Map、Filter、FlatMap的所有功能。很明显,处理函数非常强大,能够做很多之前做不到的事情。

2.2 ProcessFunction解析

  • 在源码中我们可以看到,抽象类ProcessFunction继承了AbstractRichFunction,有两个泛型类型参数:I表示 Input,也就是输入的数据类型;O表示Output,也就是处理完成之后输出的数据类型。

  • 内部单独定义了两个方法:一个是必须要实现的抽象方法.processElement();另一个是非抽象方法.onTimer()。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    @PublicEvolving
    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 {}

    …………
    }

2.2.1 抽象方法.processElement()

  • 用于“处理元素”,定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次,参数包括三个:输入数据值value,上下文ctx,以及“收集器”(Collector)out。方法没有返回值,处理之后的输出数据是通过收集器out来定义的。

    • value:当前流中的输入元素,也就是正在处理的数据,类型与流中数据类型一致。

    • ctx:类型是ProcessFunction中定义的内部抽象类Context,表示当前运行的上下文,可以获取到当前的时间戳,并提供了用于查询时间和注册定时器的“定时服务”(TimerService),以及可以将数据发送到“侧输出流”(side output)的方法.output()。Context抽象类定义如下:

      1
      2
      3
      4
      5
      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.2.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才支持设置定时器的操作,所以之前的代码中我们并没有使用定时器。所以基于不同类型的流,可以使用不同的处理函数,它们之间还是有一些微小的区别的。接下来我们就介绍一下处理函数的分类。

2.2.3 处理函数的分类

  • Flink中的处理函数其实是一个大家族,ProcessFunction只是其中一员。
  • DataStream在调用一些转换方法之后,有可能生成新的流类型;例如调用.keyBy()之后得到KeyedStream,进而再调用.window()之后得到WindowedStream。对于不同类型的流,其实都可以直接调用.process()方法进行自定义处理,这时传入的参数就都叫作处理函数。当然,它们尽管本质相同,都是可以访问状态和时间信息的底层 API,可彼此之间也会有所差异。Flink提供了8个不同的处理函数:
    • ProcessFunction:最基本的处理函数,基于DataStream直接调用.process()时作为参数传入。
    • KeyedProcessFunction:对流按键分区后的处理函数,基于KeyedStream调用.process()时作为参数传入。要想使用定时器,比如基于KeyedStream。
    • ProcessWindowFunction:开窗之后的处理函数,也是全窗口函数的代表。基于WindowedStream调用.process()时作为参数传入。
    • ProcessAllWindowFunction:同样是开窗之后的处理函数,基于AllWindowedStream调用.process()时作为参数传入。
    • CoProcessFunction:合并(connect)两条流之后的处理函数,基于ConnectedStreams调用.process()时作为参数传入。
    • ProcessJoinFunction:间隔连接(interval join)两条流之后的处理函数,基于IntervalJoined调用.process()时作为参数传入。
    • BroadcastProcessFunction:广播连接流处理函数,基于BroadcastConnectedStream调用.process()时作为参数传入。这里的“广播连接流”BroadcastConnectedStream,是一个keyBy的普通DataStream与一个广播流(BroadcastStream)做连接(conncet)之后的产物。
    • KeyedBroadcastProcessFunction:按键分区的广播连接流处理函数,同样是基于BroadcastConnectedStream调用.process()时作为参数传入。与BroadcastProcessFunction不同的是,这时的广播连接流,是一个KeyedStream与广播流(BroadcastStream)做连接之后的产物。

3、按键分区处理函数(KeyedProcessFunction)

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

3.1 定时器(Timer)和定时服务(TimerService)

  • KeyedProcessFunction的一个特色,就是可以灵活地使用定时器。

  • 定时器(timers)是处理函数中进行时间相关操作的主要机制。在.onTimer()方法中可以实现定时处理的逻辑,而它能触发的前提,就是之前曾经注册过定时器、并且现在已经到了触发时间。注册定时器的功能,是通过上下文中提供的“定时服务”(TimerService)来实现的。

  • 定时服务与当前运行的环境有关。前面已经介绍过,ProcessFunction的上下文(Context)中提供了.timerService()方法,可以直接返回一个TimerService对象:

    1
    public abstract TimerService timerService();
  • TimerService是Flink关于时间和定时器的基础服务接口,包含以下六个方法:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    // 获取当前的处理时间
    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秒一次。

    1
    2
    long coalescedTime = time / 1000 * 1000;
    ctx.timerService().registerProcessingTimeTimer(coalescedTime);
    • 这里注意定时器的时间戳必须是毫秒数,所以我们得到整秒之后还要乘以1000。定时器默认的区分精度是毫秒。
    • 另外Flink对.onTimer()和.processElement()方法是同步调用的(synchronous),所以也不会出现状态的并发修改。
  • Flink的定时器同样具有容错性,它和状态一起都会被保存到一致性检查点(checkpoint) 中。当发生故障时,Flink会重启并读取检查点中的状态,恢复定时器。如果是处理时间的定时器,有可能会出现已经“过期”的情况,这时它们会在重启时被立刻触发。

3.2 KeyedProcessFunction的使用

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

    1
    2
    stream.keyBy( t -> t.f0 )
    .process(new MyKeyedProcessFunction())
  • 类似地,KeyedProcessFunction也是继承自AbstractRichFunction的一个抽象类,源码中定义如下:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    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(),用来定义定时器触发时的回调操作。由于定时器只能在KeyedStream上使用,所以到了KeyedProcessFunction这里,我们才真正对时间有了精细的控制,定时方法.onTimer()才真正派上了用场。
  • 下面是一个使用处理时间定时器的具体示例:

    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
    public class ProcessingTimeTimerTest {
    public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // 处理时间语义,不需要分配时间戳和 watermark
    SingleOutputStreamOperator<Event> stream = env.addSource(new
    ClickSource());
    // 要用定时器,必须基于 KeyedStream
    stream.keyBy(data -> true)
    .process(new KeyedProcessFunction<Boolean, Event, String>() {
    @Override
    public void processElement(Event value, Context ctx,
    Collector<String> out) throws Exception {
    Long currTs = ctx.timerService().currentProcessingTime();
    out.collect("数据到达,到达时间:" + new Timestamp(currTs));
    // 注册一个 10 秒后的定时器
    ctx.timerService().registerProcessingTimeTimer(currTs + 10
    * 1000L);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
    Collector<String> out) throws Exception {
    out.collect("定时器触发,触发时间:" + new Timestamp(timestamp));
    }
    })
    .print();
    env.execute();
    }
    }
    1664009145293
    • 在上面的代码中,由于定时器只能在KeyedStream上使用,所以先要进行keyBy;这里的.keyBy(data -> true)是将所有数据的key都指定为了true,其实就是所有数据拥有相同的key,会分配到同一个分区。
    • 之后我们自定义了一个KeyedProcessFunction,其中.processElement()方法是每来一个数据都会调用一次,主要是定义了一个10秒之后的定时器;而.onTimer()方法则会在定时器触发时调用。所以我们会看到,程序运行后先在控制台输出“数据到达”的信息,等待10秒之后, 又会输出“定时器触发”的信息,打印出的时间间隔正是10秒。
  • 上面的例子是处理时间的定时器,所以我们是真的需要等待10秒才会看到结果。事件时间语义下,又会有什么不同呢?我们可以对上面的代码略作修改,做一个测试:

    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
    public class EventTimeTimerTest {
    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>forMonotonousTimestamps()
    .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
    @Override
    public long extractTimestamp(Event element, long
    recordTimestamp) {
    return element.timestamp;
    }
    }));

    // 基于 KeyedStream 定义事件时间定时器
    stream.keyBy(data -> true)
    .process(new KeyedProcessFunction<Boolean, Event, String>() {
    @Override
    public void processElement(Event value, Context ctx,
    Collector<String> 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<String> out) throws Exception {
    out.collect("定时器触发,触发时间:" + timestamp);
    }
    })
    .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));
    // 为了更加明显,中间停顿 5 秒钟
    Thread.sleep(5000L);
    // 发出 10 秒后的数据
    ctx.collect(new Event("Mary", "./home", 11000L));
    Thread.sleep(5000L);
    // 发出 10 秒+1ms 后的数据
    ctx.collect(new Event("Alice", "./cart", 11001L));
    Thread.sleep(5000L);
    }

    @Override
    public void cancel() {
    }
    }
    }
    • 由于是事件时间语义,所以我们必须从数据中提取出数据产生的时间戳。这里为了更清楚地看到程序行为,我们自定义了一个数据源,发出三条测试数据,时间戳分别为1000、11000和11001,并且发出数据后都会停顿5秒。

    • 在代码中,我们依然将所有数据分到同一分区,然后在自定义的KeyedProcessFunction中使用定时器。同样地,每来一条数据,我们就将当前的数据时间戳和水位线信息输出,并注册一个10秒后(以当前数据时间戳为基准)的事件时间定时器。执行程序结果如下:

      1664010797058
    • 每来一条数据,都会输出两行“数据到达”的信息,并以分割线隔开;两条数据到达的时间间隔为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)。于是所有尚未触发的定时器这时就统一触发了,我们就在控制台看到了后两个定时器的触发信息。

4、窗口处理函数

4.1 窗口处理函数的使用

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

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

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

4.2 ProcessWindowFunction解析

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

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

      • key:窗口做统计计算基于的键,也就是之前keyBy用来分区的字段。
      • context:当前窗口进行计算的上下文,它的类型就是ProcessWindowFunction内部定义的抽象类 Context。
      • elements:窗口收集到用来计算的所有数据,这是一个可迭代的集合类型。
      • out:用来发送数据输出计算结果的收集器,类型为Collector。
    • 可以明显看出,这里的参数不再是一个输入数据,而是窗口中所有数据的集合。而上下文context所包含的内容也跟其他处理函数有所差别:

      1
      2
      3
      4
      5
      6
      7
      8
      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()方法中进行显式地清除,避免内存溢出。

    • 对于窗口而言,它本身的定义就包含了一个触发计算的时间点,其实一般情况下是没有必要再去做定时操作的。如果还想要有定时器的功能,Flink也提供了另外的途径——使用窗口触发器(Trigger),在触发器中也有一个TriggerContext, 它可以起到类似TimerService的作用:获取当前时间、注册和删除定时器,另外还可以获取当前的状态。这样设计无疑会让处理流程更加清晰——定时操作也是一种“触发”,所以我们就 让所有的触发操作归触发器管,而所有处理数据的操作则归窗口函数管。

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

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

5、应用案例——Top N

  • 窗口的计算处理,在实际应用中非常常见。对于一些比较复杂的需求,如果增量聚合函数无法满足,我们就需要考虑使用窗口处理函数这样的“大招”了。
  • 网站中一个非常经典的例子,就是实时统计一段时间内的热门url。例如,需要统计最近10秒钟内最热门的两个url链接,并且每5秒钟更新一次。我们知道,这可以用一个滑动窗口来实现,而“热门度”一般可以直接用访问量来表示。于是就需要开滑动窗口收集url的访问数据,按照不同的url进行统计,而后汇总排序并最终输出前两名。这其实就是著名的“Top N”问题。
  • 很显然,简单的增量聚合可以得到url链接的访问量,但是后续的排序输出Top N就很难实现了。所以接下来我们用窗口处理函数进行实现。

5.1 使用ProcessAllWindowFunction

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

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

    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
    public class ProcessAllWindowTopN {
    public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
    .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
    .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
    @Override
    public long extractTimestamp(Event element, long
    recordTimestamp) {
    return element.timestamp;
    }
    })
    );

    // 只需要 url 就可以统计数量,所以转换成 String 直接开窗统计
    SingleOutputStreamOperator<String> result = eventStream
    .map(new MapFunction<Event, String>() {
    @Override
    public String map(Event value) throws Exception {
    return value.url;
    }
    })
    .windowAll(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))) // 开滑动窗口
    .process(new ProcessAllWindowFunction<String, String, TimeWindow>() {
    @Override
    public void process(Context context, Iterable<String> elements,
    Collector<String> out) throws Exception {
    HashMap<String, Long> urlCountMap = new HashMap<>();
    // 遍历窗口中数据,将浏览量保存到一个 HashMap 中
    for (String url : elements) {
    if (urlCountMap.containsKey(url)) {
    long count = urlCountMap.get(url);
    urlCountMap.put(url, count + 1L);
    } else {
    urlCountMap.put(url, 1L);
    }
    }
    ArrayList<Tuple2<String, Long>> mapList = new ArrayList<Tuple2<String, Long>>();
    // 将浏览量数据放入 ArrayList,进行排序
    for (String key : urlCountMap.keySet()) {
    mapList.add(Tuple2.of(key, urlCountMap.get(key)));
    }
    mapList.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();
    }
    });
    // 取排序后的前两名,构建输出结果
    StringBuilder result = new StringBuilder();
    result.append("========================================\n");
    for (int i = 0; i < 2; i++) {
    Tuple2<String, Long> temp = mapList.get(i);
    String info = "浏览量 No." + (i + 1) +
    " url:" + temp.f0 +
    " 浏览量:" + temp.f1 +
    " 窗 口 结 束 时 间 : " + new Timestamp(context.window().getEnd()) + "\n";
    result.append(info);
    }

    result.append("========================================\n");
    out.collect(result.toString());
    }
    });
    result.print();
    env.execute();
    }
    }

5.2 使用KeyedProcessFunction

  • 直接使用ProcessAllWindowFunction的方式显然不够高效。如果我们可以利用增量聚合函数的特性,每来一条数据就更新一次对应url的浏览量,那么到窗口触发计算时只需要做排序输出就可以了。

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

  • 具体实现思路就是,先按照url对数据进行keyBy分区,然后开窗进行增量聚合。这里就会发现一个问题:我们进行按键分区之后,窗口的计算就会只针对当前key有效了;也就是说, 每个窗口的统计结果中,只会有一个url的浏览量,这是无法直接用ProcessWindowFunction进行排序的。所以我们只能分成两步:先对每个url链接统计出浏览量,然后再将统计结果收集起来,排序输出最终结果。因为最后的排序还是基于每个时间窗口的,所以为了让输出的统计结果中包含窗口信息,我们可以之前定义的POJO类UrlViewCount来表示,它包含了 url、浏览量(count)以及窗口的起始结束时间。之后对UrlViewCount的处理,可以先按窗口分区,然后用KeyedProcessFunction来实现。总结处理流程如下:

    • 读取数据源;
    • 筛选浏览行为(pv);
    • 提取时间戳并生成水位线;
    • 按照url进行keyBy分区操作;
    • 开长度为1小时、步长为5分钟的事件时间滑动窗口;
    • 使用增量聚合函数AggregateFunction,并结合全窗口函数WindowFunction进行窗口聚合,得到每个 url、在每个统计窗口内的浏览量,包装成UrlViewCount;
    • 按照窗口进行keyBy分区操作;
    • 对同一窗口的统计结果数据,使用KeyedProcessFunction进行收集并排序输出。
  • 糟糕的是,这里又会带来另一个问题。最后我们用KeyedProcessFunction来收集数据做排序,这时面对的就是窗口聚合之后的数据流,而窗口已经不存在了;那到底什么时候会收集齐所有数据呢?这问题听起来似乎有些没道理。我们统计浏览量的窗口已经关闭,就说明了当前已经到了要输出结果的时候,直接输出不就行了吗?

  • 没有这么简单。因为数据流中的元素是逐个到来的,所以即使理论上我们应该“同时”收到很多url的浏览量统计结果,实际也是有先后的、只能一条一条处理。下游任务(就是我们定义的KeyedProcessFunction)看到一个url的统计结果,并不能保证这个时间段的统计数据不会再来了,所以也不能贸然进行排序输出。解决的办法,自然就是要等所有数据到齐了——这很容易让我们联想起水位线设置延迟时间的方法。这里我们也可以“多等一会儿”,等到水位线真正超过了窗口结束时间,要统计的数据就肯定到齐了。

  • 具体实现上,可以采用一个延迟触发的事件时间定时器。基于窗口的结束时间来设定延迟, 其实并不需要等太久——因为我们是靠水位线的推进来触发定时器,而水位线的含义就是“之前的数据都到齐了”。所以我们只需要设置1毫秒的延迟,就一定可以保证这一点。

  • 而在等待过程中,之前已经到达的数据应该缓存起来,我们这里用一个自定义的“列表状态”(ListState)来进行存储,如图所示。这个状态需要使用富函数类的getRuntimeContext()方法获取运行时上下文来定义,我们一般把它放在open()生命周期方法中。之后每来一个UrlViewCount,就把它添加到当前的列表状态中,并注册一个触发时间为窗口结束时间加1毫秒(windowEnd + 1)的定时器。待到水位线到达这个时间,定时器触发,我们可以保证当前窗口所有url的统计结果UrlViewCount都到齐了;于是从状态中取出进行排序输出。

    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
    public class KeyedProcessTopN {
    public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    // 从自定义数据源读取数据
    SingleOutputStreamOperator<Event> eventStream = env.addSource(new
    ClickSource())
    .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
    .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
    @Override
    public long extractTimestamp(Event element, long
    recordTimestamp) {
    return element.timestamp;
    }
    }));
    // 需要按照 url 分组,求出每个 url 的访问量
    SingleOutputStreamOperator<UrlViewCount> urlCountStream =
    eventStream.keyBy(data -> data.url)
    .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
    .aggregate(new UrlViewCountAgg(), new UrlViewCountResult());
    // 对结果中同一个窗口的统计数据,进行排序处理
    SingleOutputStreamOperator<String> result = urlCountStream.keyBy(data -> data.windowEnd).process(new TopN(2));
    result.print("result");
    env.execute();
    }

    // 自定义增量聚合
    public static class UrlViewCountAgg implements AggregateFunction<Event, Long, Long> {
    @Override
    public Long createAccumulator() {
    return 0L;
    }

    @Override
    public Long add(Event value, Long accumulator) {
    return accumulator + 1;
    }

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

    @Override
    public Long merge(Long a, Long b) {
    return null;
    }
    }

    // 自定义全窗口函数,只需要包装窗口信息
    public static class UrlViewCountResult extends ProcessWindowFunction<Long, UrlViewCount, String, TimeWindow> {
    @Override
    public void process(String url, Context context, Iterable<Long> elements, Collector<UrlViewCount> out) throws Exception {
    // 结合窗口信息,包装输出内容
    Long start = context.window().getStart();
    Long end = context.window().getEnd();
    out.collect(new UrlViewCount(url, elements.iterator().next(), start, end));
    }
    }

    // 自定义处理函数,排序取 top n
    public static class TopN extends KeyedProcessFunction<Long, UrlViewCount, String> {
    // 将 n 作为属性
    private Integer n;
    // 定义一个列表状态
    private ListState<UrlViewCount> urlViewCountListState;

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

    @Override
    public void open(Configuration parameters) throws Exception {
    // 从环境中获取列表状态句柄
    urlViewCountListState = getRuntimeContext().getListState(new ListStateDescriptor<UrlViewCount>("url-view-count-list", Types.POJO(UrlViewCount.class)));
    }

    @Override
    public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception {
    // 将 count 数据添加到列表状态中,保存起来
    urlViewCountListState.add(value);
    // 注册 window end + 1ms 后的定时器,等待所有数据到齐开始排序
    ctx.timerService().registerEventTimeTimer(ctx.getCurrentKey() + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
    // 将数据从列表状态变量中取出,放入 ArrayList,方便排序
    ArrayList<UrlViewCount> urlViewCountArrayList = new ArrayList<>();
    for (UrlViewCount urlViewCount : urlViewCountListState.get()) {
    urlViewCountArrayList.add(urlViewCount);
    }
    // 清空状态,释放资源
    urlViewCountListState.clear();
    // 排序
    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("========================================\n");
    result.append("窗口结束时间:" + new Timestamp(timestamp - 1) + "\n");
    for (int i = 0; i < this.n; i++) {
    UrlViewCount UrlViewCount = urlViewCountArrayList.get(i);
    String info = "No." + (i + 1) + " "
    + "url:" + UrlViewCount.url + " "
    + "浏览量:" + UrlViewCount.count + "\n";
    result.append(info);
    }
    result.append("========================================\n");
    out.collect(result.toString());
    }
    }
    }
    1664025958750
    • 代码中,我们还利用了定时器的特性:针对同一key、同一时间戳会进行去重。所以对于同一个窗口而言,我们接到统计结果数据后设定的windowEnd + 1的定时器都是一样的,最终只会触发一次计算。而对于不同的key(这里key是windowEnd),定时器和状态都是独立的,所以我们也不用担心不同窗口间数据的干扰。

6、侧输出流(Side Output)

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

  • 之前讲到的绝大多数转换算子,输出的都是单一流,流里的数据类型只能有一种。而侧输出流可以认为是“主流”上分叉出的“支流”,所以可以由一条流产生出多条流,而且这些流中的数据类型还可以不一样。利用这个功能可以很容易地实现“分流”操作。

  • 具体应用时,只要在处理函数的.processElement()或者.onTimer()方法中,调用上下文的.output()方法就可以了。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    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));
    }
    });
    • 这里output()方法需要传入两个参数,第一个是一个“输出标签”OutputTag,用来标识侧输出流,一般会在外部统一声明;第二个就是要输出的数据。我们可以在外部先将OutputTag声明出来:

      1
      OutputTag<String> outputTag = new OutputTag<String>("side-output") {};
    • 如果想要获取这个侧输出流,可以基于处理之后的DataStream直接调用.getSideOutput()方法,传入对应的OutputTag,这个方式与窗口API中获取侧输出流是完全一样的。

      1
      DataStream<String> stringStream = longStream.getSideOutput(outputTag);