Flink处理函数(ProcessFunction、KeyedProcessFunction、ProcessWindowFunction、 ProcessAllWindowFunction)

刺骨的言语ヽ痛彻心扉 2024-04-22 07:01 40阅读 0赞

#### 文章目录 ####

*  处理函数
 *   *  基本处理函数（ProcessFunction）
     *   *  ProcessFunction简述以及使用
         *  ProcessFunction 解析
     *  按键分区处理函数（KeyedProcessFunction）
     *   *  定时器（Timer）和定时服务（TimerService）
         *  KeyedProcessFunction 的使用
         *   *  自定义数据类
             *  自定义处理函数
             *  自定义函数使用
     *  窗口处理函数ProcessWindowFunction
     *   *  ProcessWindowFunction 解析
         *  ProcessWindowFunction 使用
         *   *  自定义数据类
             *  自定义函数
             *  自定义函数使用
     *  全窗口处理函数ProcessAllWindowFunction
     *   *  ProcessAllWindowFunction简述
         *  ProcessAllWindowFunction使用
         *   *  自定义数据类
             *  自定义函数
             *  自定义函数使用

## 处理函数 ##

DataStream API，这也是 Flink 编程的核心。而我们知道，为了让代码有更强大的表现力和易用性，Flink 本身提供了多层 API，DataStream API 只是中间的一环,如下图所示:

![在这里插入图片描述][c5ac784cabe946b0a41567bd2b719715.png]  
在更底层，我们可以不定义任何具体的算子（比如 map，filter，或者 window），而只是提炼出一个统一的“处理”（process）操作——它是所有转换算子的一个概括性的表达，可以自定义处理逻辑，所以这一层接口就被叫作“处理函数”（process function）。在处理函数中，我们直面的就是数据流中最基本的元素：数据事件（event）、状态（state）以及时间（time）。这就相当于对流有了完全的控制权。处理函数比较抽象，没有具体的操作，所以对于一些常见的简单应用（比如求和、开窗口）会显得有些麻烦；不过正是因为它不限定具体做什么，所以理论上我们可以做任何事情，实现所有需求。

**接下来我们简单了解下相关处理函数ProcessFunction、KeyedProcessFunction、ProcessFunction以及ProcessAllWindowFunction**

?在Apache Flink中，ProcessFunction是用于处理数据流的通用函数。它是一个抽象类，定义了处理数据流的常用方法，如processElement，onTimer等。您可以扩展ProcessFunction类并重写这些方法，以便在Flink程序中执行复杂的数据流处理逻辑。

?KeyedProcessFunction是ProcessFunction的特殊类型，用于处理带有键的数据流。它定义了额外的方法，如getKey，context.timerService()等，用于访问数据流中每个元素的键以及在处理函数中安排定时器。

?ProcessWindowFunction和ProcessAllWindowFunction是用于处理时间窗口的特殊函数。它们提供了一个process方法，用于在每个窗口中对数据进行处理。ProcessWindowFunction接受带有键的数据流，并且每个窗口都对应于一个键，而ProcessAllWindowFunction接受不带键的数据流，并且每个窗口都包含整个数据流。

### 基本处理函数（ProcessFunction） ###

#### ProcessFunction简述以及使用 ####

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

stream.process(new MyProcessFunction())

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

代码简述  
**假设你有一个DataStream，其中包含单词的字符串。你想要使用ProcessFunction统计每个单词的出现次数。你可以这样写：**

DataStream<String> inputStream = …;
    
    DataStream<Tuple2<String, Long>> outputStream = inputStream
        .keyBy(word -> word)
        .process(new WordCountProcessFunction());
    
    class WordCountProcessFunction extends ProcessFunction<String, Tuple2<String, Long>> {
        
      private MapState<String, Long> countState;
    
      @Override
      public void open(Configuration parameters) throws Exception {
        
        countState = getRuntimeContext().getMapState(new MapStateDescriptor<>("counts", String.class, Long.class));
      }
    
      @Override
      public void processElement(String word, Context context, Collector<Tuple2<String, Long>> collector) throws Exception {
        
        Long count = countState.get(word);
        if (count == null) {
        
          count = 0L;
        }
        countState.put(word, count + 1);
        collector.collect(new Tuple2<>(word, count + 1));
      }
    }

这里我们在 ProcessFunction 中重写了.processElement()方法，自定义了一种处理逻辑

#### ProcessFunction 解析 ####

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

public abstract class ProcessFunction<I, O> extends AbstractRichFunction {
        
    ...
    public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;
    
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {
        }
    ...
    }

1.  抽象方法.processElement()  
    用于“处理元素”，定义了处理的核心逻辑。这个方法对于流中的每个元素都会调用一次，参数包括三个：输入数据值 value，上下文 ctx，以及“收集器”（Collector）out。方法没有返回值，处理之后的输出数据是通过收集器 out 来定义的。  
    ⚫ value：当前流中的输入元素，也就是正在处理的数据，类型与流中数据类  
    型一致。  
    ⚫ ctx：类型是 ProcessFunction 中定义的内部抽象类 Context，表示当前运行的上下文，可以获取到当前的时间戳，并提供了用于查询时间和注册定时器的“定时服务”(TimerService)，以及可以将数据发送到“侧输出流”（side output）的方法.output()。  
    Context 抽象类定义如下：

public abstract class Context {
        
    public abstract Long timestamp();
    public abstract TimerService timerService();
    public abstract <X> void output(OutputTag<X> outputTag, X value);
    }

⚫ out：“收集器”（类型为 Collector），用于返回输出数据。使用方式与 flatMap算子中的收集器完全一样，直接调用 out.collect()方法就可以向下游发出一个数据。  
这个方法可以多次调用，也可以不调用。通过几个参数的分析不难发现，ProcessFunction 可以轻松实现 flatMap 这样的基本转换功能（当然 map、filter 更不在话下）；而通过富函数提供的获取上下文方法.getRuntimeContext()，也可以自定义状态（state）进行处理，这也就能实现聚合操作的功能了。

1.  非抽象方法.onTimer()  
    用于定义定时触发的操作，这是一个非常强大、也非常有趣的功能。这个方法只有在注册好的定时器触发的时候才会调用，而定时器是通过“定时服务”TimerService 来注册的。打个比方，注册定时器（timer）就是设了一个闹钟，到了设定时间就会响；而.onTimer()中定义的，就是闹钟响的时候要做的事。所以它本质上是一个基于时间的“回调”（callback）方法，通过时间的进展来触发；在事件时间语义下就是由水位线（watermark）来触发了。与.processElement()类似，定时方法.onTimer()也有三个参数：时间戳（timestamp），上下文（ctx），以及收集器（out）。这里的 timestamp 是指设定好的触发时间，事件时间语义下当然就是水位线了。另外这里同样有上下文和收集器，所以也可以调用定时服务（TimerService），以及任意输出处理之后的数据。  
    既然有.onTimer()方法做定时触发，我们用 ProcessFunction 也可以自定义数据按照时间分组、定时触发计算输出结果；这其实就实现了窗口（window）的功能。所以说 ProcessFunction是真正意义上的终极奥义，用它可以实现一切功能。  
    我们也可以看到，处理函数都是基于事件触发的。水位线就如同插入流中的一条数据一样；只不过处理真正的数据事件调用的是.processElement()方法，而处理水位线事件调用的是.onTimer()。  
    这里需要注意的是，上面的.onTimer()方法只是定时器触发时的操作，而定时器（timer）真正的设置需要用到上下文 ctx 中的定时服务。在 Flink 中，只有“按键分区流”KeyedStream才支持设置定时器的操作，所以之前的代码中我们并没有使用定时器。所以基于不同类型的流，

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

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

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

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

> public abstract TimerService timerService();

**TimerService 是 Flink 关于时间和定时器的基础服务接口，包含以下六个方法：**

> // 获取当前的处理时间  
> long currentProcessingTime();  
> // 获取当前的水位线（事件时间）  
> long currentWatermark();  
> // 注册处理时间定时器，当处理时间超过 time 时触发  
> void registerProcessingTimeTimer(long time);  
> // 注册事件时间定时器，当水位线超过 time 时触发  
> void registerEventTimeTimer(long time);  
> // 删除触发时间为 time 的处理时间定时器  
> void deleteProcessingTimeTimer(long time);  
> // 删除触发时间为 time 的处理时间定时器  
> void deleteEventTimeTimer(long time);

六个方法可以分成两大类：基于处理时间和基于事件时间。而对应的操作主要有三个：获取当前时间，注册定时器，以及删除定时器。需要注意，尽管处理函数中都可以直接访问TimerService，不过只有基于 KeyedStream 的处理函数，才能去调用注册和删除定时器的方法；  
未作按键分区的 DataStream 不支持定时器操作，只能获取当前时间。  
对于处理时间和事件时间这两种类型的定时器，TimerService 内部会用一个优先队列将它们的时间戳（timestamp）保存起来，排队等待执行。可以认为，定时器其实是 KeyedStream上处理算子的一个状态，它以时间戳作为区分。所以 TimerService 会以键（key）和时间戳为标准，对定时器进行去重；也就是说对于每个 key 和时间戳，最多只有一个定时器，如果注册了多次，onTimer()方法也将只被调用一次。这样一来，我们在代码中就方便了很多，可以肆无忌惮地对一个 key 注册定时器，而不用担心重复定义——因为一个时间戳上的定时器只会触发一次。  
基于 KeyedStream 注册定时器时，会传入一个定时器触发的时间戳，这个时间戳的定时器  
对于每个 key 都是有效的。这样，我们的代码并不需要做额外的处理，底层就可以直接对不同key 进行独立的处理操作了。利用这个特性，有时我们可以故意降低时间戳的精度，来减少定时器的数量，从而提高处理性能。比如我们可以在设置定时器时只保留整秒数，那么定时器的触发频率就是最多 1 秒一次。

> long coalescedTime = time / 1000 \* 1000;  
> ctx.timerService().registerProcessingTimeTimer(coalescedTime);

#### KeyedProcessFunction 的使用 ####

KeyedProcessFunction 也是继承自 AbstractRichFunction 的一个抽象类，源码中定义如下：

public abstract class KeyedProcessFunction<K, I, O> extends AbstractRichFunction 
    {
        
    ...
    public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception;
    
    public void onTimer(long timestamp, OnTimerContext ctx,Collector<O> out) throws Exception {
        }
    public abstract class Context {
        ...}
    ...
    }

可以看到与 ProcessFunction 的定义几乎完全一样，区别只是在于类型参数多了一个 K，这是当前按键分区的 key 的类型。同样地，我们必须实现一个.processElement()抽象方法，用来处理流中的每一个数据；另外还有一个非抽象方法.onTimer()，用来定义定时器触发时的回调操作。由于定时器只能在 KeyedStream 上使用，所以到了 KeyedProcessFunction 这里，我们才真正对时间有了精细的控制，定时方法.onTimer()才真正派上了用场。

**假设你有一个流数据流，其中包含每个用户的点击数据，并且你想要对每个用户的点击数进行计数。你可以使用 KeyedProcessFunction 来实现这个功能，如下所示：**

##### 自定义数据类 #####

public class ClickEvent {
        
      private String userId;
      private long timestamp;
    
      // constructor, getters and setters
    }

##### 自定义处理函数 #####

public class ClickCountProcessFunction extends KeyedProcessFunction<String, ClickEvent, Tuple2<String, Long>> {
        
    
      private ValueState<Long> clickCountState;
    
      @Override
      public void open(Configuration parameters) throws Exception {
        
        super.open(parameters);
        clickCountState = getRuntimeContext().getState(new ValueStateDescriptor<>("click-count", Long.class));
      }
    
      @Override
      public void processElement(ClickEvent value, Context ctx, Collector<Tuple2<String, Long>> out) throws Exception {
        
        long count = clickCountState.value() + 1;
        clickCountState.update(count);
        out.collect(Tuple2.of(value.getUserId(), count));
      }
    }

##### 自定义函数使用 #####

DataStream<ClickEvent> clickEventStream = ...;
    
    DataStream<Tuple2<String, Long>> clickCountStream = clickEventStream
        .keyBy(ClickEvent::getUserId)
        .process(new ClickCountProcessFunction());

### 窗口处理函数ProcessWindowFunction ###

#### ProcessWindowFunction 解析 ####

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

public abstract class ProcessWindowFunction<IN, OUT, KEY, W extends Window>
    extends AbstractRichFunction {
        
    ...
    public abstract void process(KEY key, Context context, Iterable<IN> elements, Collector<OUT> out) throws Exception;
    
    public void clear(Context context) throws Exception {
        }
    
    public abstract class Context implements java.io.Serializable {
        ...}
    }

**ProcessWindowFunction 依然是一个继承了 AbstractRichFunction 的抽象类，它有四个类型**  
参数：  
⚫ IN：input，数据流中窗口任务的输入数据类型。  
⚫ OUT：output，窗口任务进行计算之后的输出数据类型。  
⚫ KEY：数据中键 key 的类型。  
⚫ W：窗口的类型，是 Window 的子类型。一般情况下我们定义时间窗口，W就是 TimeWindow。  
 **而内部定义的方法，跟我们之前熟悉的处理函数就有所区别了。因为全窗口函数不是逐个处理元素的，所以处理数据的方法在这里并不是.processElement()，而是改成了.process()。方法包含四个参数。**  
⚫ key：窗口做统计计算基于的键，也就是之前 keyBy 用来分区的字段。  
⚫ context：当前窗口进行计算的上下文，它的类型就是 ProcessWindowFunction  
内部定义的抽象类 Context。 ⚫ elements：窗口收集到用来计算的所有数据，这是一个可迭代的集合类型。  
⚫ out：用来发送数据输出计算结果的收集器，类型为 Collector。  
可以明显看出，这里的参数不再是一个输入数据，而是窗口中所有数据的集合。而上下文context 所包含的内容也跟其他处理函数有所差别：

public abstract class Context implements java.io.Serializable {
        
    	public abstract W window();
    	public abstract long currentProcessingTime();
    	public abstract long currentWatermark();
    	public abstract KeyedStateStore windowState();
    	public abstract KeyedStateStore globalState();
    	public abstract <X> void output(OutputTag<X> outputTag, X value);
    }

#### ProcessWindowFunction 使用 ####

假设你有一个流数据流，其中包含每个用户的点击数据，并且你想要对每个用户在每小时内的点击数进行计数。你可以使用 ProcessWindowFunction 来实现这个功能，如下所示：

##### 自定义数据类 #####

public class ClickEvent {
        
      private String userId;
      private long timestamp;
    
      // constructor, getters and setters
    }

##### 自定义函数 #####

public class ClickCountWindowFunction
        extends ProcessWindowFunction<ClickEvent, Tuple2<String, Long>, String, TimeWindow> {
        
    
      @Override
      public void process(String userId,
                          Context context,
                          Iterable<ClickEvent> events,
                          Collector<Tuple2<String, Long>> out) {
        
        long count = 0L;
        for (ClickEvent event : events) {
        
          count++;
        }
        out.collect(Tuple2.of(userId, count));
      }
    }

##### 自定义函数使用 #####

DataStream<ClickEvent> clickEventStream = ...;
    
    DataStream<Tuple2<String, Long>> clickCountStream = clickEventStream
        .keyBy(ClickEvent::getUserId)
        //定义一小时窗口
        .timeWindow(Time.hours(1))
        .process(new ClickCountWindowFunction());

### 全窗口处理函数ProcessAllWindowFunction ###

#### ProcessAllWindowFunction简述 ####

它与 ProcessWindowFunction 类似，但是它会对窗口中的所有数据进行处理，而不是仅处理触发窗口计算的数据。

#### ProcessAllWindowFunction使用 ####

##### 自定义数据类 #####

public class ClickEvent {
        
      private String userId;
      private long timestamp;
    
      // constructor, getters and setters
    }

##### 自定义函数 #####

public class ClickCountProcessAllWindowFunction
        extends ProcessAllWindowFunction<ClickEvent, Tuple2<String, Long>, TimeWindow> {
        
    
      @Override
      public void process(Context context,
                          Iterable<ClickEvent> events,
                          Collector<Tuple2<String, Long>> out) {
        
        long count = 0L;
        String userId = null;
        for (ClickEvent event : events) {
        
          count++;
          userId = event.getUserId();
        }
        out.collect(Tuple2.of(userId, count));
      }
    }

##### 自定义函数使用 #####

DataStream<ClickEvent> clickEventStream = ...;
    
    DataStream<Tuple2<String, Long>> clickCountStream = clickEventStream
        .keyBy(ClickEvent::getUserId)
        .timeWindow(Time.hours(1))
        .process(new ClickCountProcessAllWindowFunction());

上面的代码示例是使用 ProcessAllWindowFunction 对每个用户在一小时内的点击数进行计数的例子。

首先，我们对点击事件流使用了 keyBy 操作来对事件进行分组，然后使用 timeWindow 操作将每个用户的点击事件分到一个一小时的窗口中。然后，我们使用自定义的 ClickCountProcessAllWindowFunction 对窗口中的所有数据进行处理，计算每个用户在一小时内的点击数。

最后，我们使用 process 操作将 ClickCountProcessAllWindowFunction 应用到数据流上，并将处理后的结果输出到一个新的流中。这个新的流中的每个元素都是一个包含用户 ID 和点击数的元组，例如 Tuple2<String, Long>。

[c5ac784cabe946b0a41567bd2b719715.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/22/ef19b5cd95b440b6a441160ee7e0a7c9.png