当前位置:   article > 正文

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

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

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

处理函数

处理函数是最为灵活的处理方法,可以实现各种自定义的业务逻辑;同时也是整个 DataStream API 的底层基础。

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

  • ProcessFunction是用于处理数据流的通用函数。它是一个抽象类,定义了处理数据流的常用方法,如processElement,onTimer等。您可以扩展ProcessFunction类并重写这些方法,以便在Flink程序中执行复杂的数据流处理逻辑。
  • KeyedProcessFunction是ProcessFunction的特殊类型,用于处理带有键的数据流。它定义了额外的方法,如getKey,context.timerService()等,用于访问数据流中每个元素的键以及在处理函数中安排定时器。
  • ProcessWindowFunction和ProcessAllWindowFunction是用于处理时间窗口的特殊函数。它们提供了一个process方法,用于在每个窗口中对数据进行处理。ProcessWindowFunction接受带有键的数据流,并且每个窗口都对应于一个键,而ProcessAllWindowFunction接受不带键的数据流,并且每个窗口都包含整个数据流。

基本处理函数(ProcessFunction)

提供了一个“定时服务”(TimerService),我们可以通过它访问流中的事件(event)、时间戳(timestamp)、水位线(watermark),甚至可以注册“定时事件”。而且处理函数继承了 AbstractRichFunction 抽象类,所以拥有富函数类的所有特性,同样可以访问状态(state)和其他运行时信息。此外,处理函数还可以直接将数据输出到侧输出流(side output)中。

基于 DataStream 调用.process()方法就可以了。方法需要传入一个 ProcessFunction 作为参数,用来定义处理逻辑。

stream.process(new MyProcessFunction())

ProcessFunction 是一个抽象类,继承了AbstractRichFunction;MyProcessFunction 是它的一个具体实现。所以所有的处理函数,都是富函数(RichFunction),富函数可以调用的东西这里同样都可以调用。

假设你有一个DataStream,其中包含单词的字符串。你想要使用ProcessFunction统计每个单词的出现次数。

  1. DataStream<String> inputStream = …;
  2. DataStream<Tuple2<String, Long>> outputStream = inputStream
  3. .keyBy(word -> word)
  4. .process(new WordCountProcessFunction());
  5. class WordCountProcessFunction extends ProcessFunction<String, Tuple2<String, Long>> {
  6. private MapState<String, Long> countState;
  7. @Override
  8. public void open(Configuration parameters) throws Exception {
  9. countState = getRuntimeContext().getMapState(new MapStateDescriptor<>("counts", String.class, Long.class));
  10. }
  11. @Override
  12. public void processElement(String word, Context context, Collector<Tuple2<String, Long>> collector) throws Exception {
  13. Long count = countState.get(word);
  14. if (count == null) {
  15. count = 0L;
  16. }
  17. countState.put(word, count + 1);
  18. collector.collect(new Tuple2<>(word, count + 1));
  19. }
  20. }

抽象类 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 {}
...
}

非抽象方法.onTimer()

用于定义定时触发的操作,这是一个非常强大、也非常有趣的功能。这个方法只有在注册好的定时器触发的时候才会调用,而定时器是通过“定时服务”TimerService 来注册的。

按键分区处理函数(KeyedProcessFunction)

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

定时器(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);

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 {...}
...
}

K,这是当前按键分区的 key 的类型

假设你有一个流数据流,其中包含每个用户的点击数据,并且你想要对每个用户的点击数进行计数。

自定义数据类
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

它继承了 AbstractRichFunction 的抽象类,它有四个类型参数:

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

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

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

假设你有一个流数据流,其中包含每个用户的点击数据,并且你想要对每个用户在每小时内的点击数进行计数。

自定义数据类
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

自定义数据类
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());
声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/凡人多烦事01/article/detail/78408
推荐阅读
相关标签
  

闽ICP备14008679号