当前位置: 首页 > news >正文

【Flink】Flink 处理函数之基本处理函数(一)

1. 处理函数介绍

流处理API,无论是基本的转换聚合、还是复杂的窗口操作,都是基于DataStream进行转换的,所以统称为DataStreamAPI,这是Flink编程的核心。

但其实Flink为了更强大的表现力和易用性,Flink本身提供了多层API,DataStreamAPI只是中间一环,如下图所示:在这里插入图片描述
在更底层,Flink可以不定义任何具体的算子(比如 mapfilter,或者 window),而只是提炼出一个统一的“处理”(process)操作——它是所有转换算子的一个概括性的表达,可以自定义处理逻辑,所以这一层接口就被叫作“处理函数”(process function)

在处理函数中,操作的就是数据流中最基本的元素:数据事件(event)状态(state)以及时间(time)。这就相当于对流有了完全的控制权。处理函数比较抽象,没有具体的操作,所以对于一些常见的简单应用(比如求和、开窗口)会显得有些麻烦;不过正是因为它不限定具体做什么,所以理论上我们可以做任何事情,实现所有需求。

2. 处理函数的分类

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

2.1 基本处理函数(ProcessFunction)

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

2.1.1 处理函数的功能和使用

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

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

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

stream.process(new MyProcessFunction())

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

代码实例:

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>() {@Overridepublic long extractTimestamp(Event element, long recordTimestamp) {return element.getTimestamp();}}));stream.process(new ProcessFunction<Event, String>() {@Overridepublic void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {out.collect(value.toString());if (value.getUser().equals("Mary")) {out.collect(value.user + "click " + value.getUrl());} else if (value.getUser().equals("Alice")) {out.collect(value.user);out.collect(value.user);}System.out.println("timestamp:" + ctx.timestamp());System.out.println("watermark:" + ctx.timerService().currentWatermark());System.out.println(getRuntimeContext().getIndexOfThisSubtask());}@Overridepublic void onTimer(long timestamp, ProcessFunction<Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {super.onTimer(timestamp, ctx, out);}@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);}@Overridepublic void close() throws Exception {super.close();}}).print();env.execute();}

运行结果:
在这里插入图片描述

这里第一次的水位线的值其实是个默认值,Long.MIN_VALUE + outOfOrdernessMillis + 1;
在这里插入图片描述

然后每次下一次的水位线都是上一次的timestamp - 1

2.1.2 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 {}
...
}
2.1.2.1 抽象方法.processElement()

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

  • value: 当前流中的输入元素,也就是正在处理的数据,类型与流中数据类型一致。
  • cts:类型是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这样的基本转换功能(当然 mapfilter 更不在话下);而通过富函数提供的获取上下文方法.getRuntimeContext(),也可以自定义状态(state)进行处理,这也就能实现聚合操作的功能了。

2.1.2.2 非抽象方法.onTimer()
@Override
public void onTimer(long timestamp, ProcessFunction<Event, String>.OnTimerContext ctx, Collector<String> out) throws Exception {super.onTimer(timestamp, ctx, out);
}

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

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

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

处理函数都是基于事件触发的。水位线就如同插入流中的一条数据一样;只不过处理真正的数据事件调用的是.processElement()方法,而处理水位线事件调用的是.onTimer()

.onTimer()方法只是定时器触发时的操作,而定时器(timer)真正的设置需要用到上下文 ctx 中的定时服务。在Flink中,只有“按键分区流”KeyedStream才支持设置定时器的操作,所以之前的代码中并没有使用定时器。所以基于不同类型的流,可以使用不同的处理函数。

相关文章:

  • STM32-01基于HAL库(CubeMX+MDK+Proteus)仿真开发环境搭建(LED点亮测试实例)
  • Redis 教程系列之Redis 事务(十六)
  • 前端理论总结(css3)——css优化的方法
  • 线程通信同步
  • 低代码与数字化转型:重塑企业技术生态的新引擎
  • STM32之HAL开发——串口配置(源码)
  • mac系统使用经验
  • Etcd 基本入门
  • Pandas合并数据集
  • 线程安全(买票案例):加锁方式(synchronized、Lock锁)【同步代码块、同步方法】--学习JavaEE的day31上
  • ROS2高效学习第十章 -- ros2 高级组件之 component 合并进程启动 其一
  • 图像分类实战:深度学习在CIFAR-10数据集上的应用
  • Java 中文官方教程 2022 版翻译完成
  • LQR的横向控制与算法仿真实现
  • BaseDao封装增删改查
  • Javascript基础之Array数组API
  • JAVA多线程机制解析-volatilesynchronized
  • SegmentFault 2015 Top Rank
  • SegmentFault 技术周刊 Vol.27 - Git 学习宝典:程序员走江湖必备
  • STAR法则
  • vue--为什么data属性必须是一个函数
  • 后端_MYSQL
  • 如何利用MongoDB打造TOP榜小程序
  • 体验javascript之美-第五课 匿名函数自执行和闭包是一回事儿吗?
  • 我这样减少了26.5M Java内存!
  • 小程序开发中的那些坑
  • NLPIR智能语义技术让大数据挖掘更简单
  • 宾利慕尚创始人典藏版国内首秀,2025年前实现全系车型电动化 | 2019上海车展 ...
  • ​2020 年大前端技术趋势解读
  • (k8s中)docker netty OOM问题记录
  • (附源码)ssm基于微信小程序的疫苗管理系统 毕业设计 092354
  • (附源码)ssm智慧社区管理系统 毕业设计 101635
  • (简单) HDU 2612 Find a way,BFS。
  • (十一)JAVA springboot ssm b2b2c多用户商城系统源码:服务网关Zuul高级篇
  • (转)关于多人操作数据的处理策略
  • (转)一些感悟
  • (轉貼) 資訊相關科系畢業的學生,未來會是什麼樣子?(Misc)
  • ***php进行支付宝开发中return_url和notify_url的区别分析
  • .net 4.0发布后不能正常显示图片问题
  • .NET 8.0 发布到 IIS
  • .NET Conf 2023 回顾 – 庆祝社区、创新和 .NET 8 的发布
  • .NET Core 实现 Redis 批量查询指定格式的Key
  • .NET设计模式(2):单件模式(Singleton Pattern)
  • .net专家(高海东的专栏)
  • /usr/bin/env: node: No such file or directory
  • @TableId注解详细介绍 mybaits 实体类主键注解
  • [ 环境搭建篇 ] 安装 java 环境并配置环境变量(附 JDK1.8 安装包)
  • []我的函数库
  • [2009][note]构成理想导体超材料的有源THz欺骗表面等离子激元开关——
  • [bzoj1901]: Zju2112 Dynamic Rankings
  • [caffe(二)]Python加载训练caffe模型并进行测试1
  • [EULAR文摘] 脊柱放射学持续进展是否显著影响关节功能
  • [Hive] CTE 通用表达式 WITH关键字
  • [Java] 模拟Jdk 以及 CGLib 代理原理
  • [LeetCode] Longest Common Prefix 字符串公有前序