Flink学习笔记(二)Flink常用API详解
创始人
2024-06-02 14:42:09
0

一、Flink API的类型:

1.低级api:

        提供了对时间和状态的细粒度控制,简洁性和易用性较差,主要应用在对一些复杂事件的处理逻辑上。

2.核心api:

        要提供了针对流数据和离线数据的处理,对低级API进行了一些封装,提供了filter、sum、max、min等高级函数,简单且易用,所以在工作中应用比较广泛。核心api分成两类

(1)DataStream API:用于处理无界数据流,提供了各种操作符来处理流数据。

(2)DataSet API:用于处理有界数据集,提供了各种操作符来处理批处理数据。

3.Table api/sql:

(1)Table API:一般与DataSet或者DataStream紧密关联,首先通过一个DataSet或DataStream创建出一个Table;然后用类似于filter、join或者select关系型转化操作来转化为一个新的Table对象;最后将一个Table对象转回一个DataSet或DataStream。与SQL不同的是,Table API的查询不是一个指定的SQL字符串,而是调用指定的API方法

(2)SQL:Flink的SQL集成是基于Apache Calcite的,Apache Calcite实现了标准的SQL,使用起来比其他API更加灵活,因为可以直接使用SQL语句。Table API和SQL可以很容易地结合在一块使用,它们都返回Table对象

二、Flink DataStream的常用API:

DataStream API主要分为3块:DataSource、Transformation、Sink。

1.DataSource 输入源:

Flink针对DataStream提供了大量的已经实现的DataSource(数据源)接口,比如下面4种。

(1)基于文件:读取文本文件,文件遵循TextInputFormat逐行读取规则并返回

(2)基于Socket:从Socket中读取数据,元素可以通过一个分隔符分开

(3)基于集合:通过Java的Collection集合创建一个数据流,集合中的所有元素必须是相同类型的

(4)自定义输入:addSource可以实现读取第三方数据源的数据。Flink也提供了一批内置的Connector(连接器)。连接器会提供对应的Source支持,如mq(kafka/RabbitMQ)、es、redis、mysql(通过JDBC连接器)等等

2.Transformation 转换器:

        它对一个或多个输入数据源进行计算处理,比如Map、FlatMap和Filter等操作,Flink针对DataStream提供了大量的已经实现的算子:

(1)Map:输入一个元素,然后返回一个元素,中间可以进行清洗转换等操作。

(2)FlatMap:输入一个元素,可以返回零个、一个或者多个元素。

(3)Filter:过滤函数,对传入的数据进行判断,符合条件的数据会被留下。

(4)KeyBy:根据指定的Key进行分组,Key相同的数据会进入同一个分区。

        KeyBy的两种典型用法如下:
                DataStream.keyBy("someKey") 指定对象中的someKey段作为分组Key。
                DataStream.keyBy(0) 指定Tuple中的第一个元素作为分组Key

(5)Reduce:对数据进行聚合操作,结合当前元素和上一次Reduce返回的值进行聚合操作,然后返回一个新的值

(6)Aggregations:sum()、min()、max()等

(7)Union:合并多个流,新的流会包含所有流中的数据,但是Union有一个限制,就是所有合并的流类型必须是一致的

(8)Connect:和Union类似,但是只能连接两个流,两个流的数据类型可以不同,会对两个流中的数据应用不同的处理方法。

(9)coMap和coFlatMap:在ConnectedStream中需要使用这种函数,类似于Map和flatMap

(10)Split:根据规则把一个数据流切分为多个流

(11)Select:和Split配合使用,选择切分后的流

3.Sink 输出源:

Flink针对DataStream提供了大量的已经实现的数据目的地(Sink)

(1)writeAsText():将元素以字符串形式逐行写入,这些字符串通过调用每个元素的toString()方法来获取。

(2)print() / printToErr():打印每个元素的toString()方法的值到标准输出或者标准错误输出流中。

(3)自定义输出:addSink可以实现把数据输出到第三方存储介质中。如hdfs、mysql(通过JDBC连接器)、es、kafka、redis

三、Flink DataSet的常用API分析

DataSet API也可以分为3块来分析:DataSource、Transformation和Sink,使用类似,这里只说对比DataStream的区别:

  1. 数据类型:DataSet API适用于处理有界数据,即离线批处理;DataStream API适用于处理无界数据,即实时流处理。

  2. 数据处理方式:DataSet API采用批处理方式,即将数据读取到内存中,进行批量计算,然后将结果写出;DataStream API采用流处理方式,即从数据流中逐个读取数据,进行实时计算,然后将结果发送到下游。

  3. 时间处理:在DataStream API中,时间处理非常重要,可以通过时间窗口、时间滑动窗口等方式对数据进行处理。而在DataSet API中,时间处理相对简单,通常只需要使用时间戳进行排序和分组即可。

  4. 窗口:DataStream API中支持各种窗口,如滚动窗口、滑动窗口、会话窗口等,而DataSet API中则不支持窗口。

  5. 稳定性:由于DataStream API中处理的是实时数据流,因此对数据的稳定性要求较高,需要考虑数据丢失、重复等问题。而DataSet API中处理的是静态数据,不需要考虑这些问题。

四、Table API和SQL的基本使用

        Flink针对标准的流处理和批处理提供了两种关系型API:Table API和SQL。Table API允许用户以一种很直观的方式进行select、filter和join操作;Flink SQL支持基于 ApacheCalcite实现的标准SQL。针对批处理和流处理可以提供相同的处理语义和结果。

        Table API和SQL是关系型API,用户可以像操作MySQL数据库表一样来操作数据,而不需要通过编写Java代码来完成Flink Function,更不需要手工为Java代码调优。另外,SQL作为一个非程序员可操作的语言,学习成本很低,如果一个系统提供SQL支持,将很容易被用户接受。

1.Table API的基本使用

(1)创建TableEnvironment对象

TableEnvironment是Flink Table API的主要入口,它提供了各种方法来创建Table对象、注册表、执行查询等操作。可以通过以下方式创建TableEnvironment对象:

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = StreamTableEnvironment.create(env)

(2)创建Table对象

在Table API中,可以通过fromDataStream()方法将DataStream转换为Table,也可以通过其他方法创建Table对象。例如,以下代码创建了一个Table对象:

val table = tEnv.fromValues(DataTypes.ROW(DataTypes.FIELD("name", DataTypes.STRING()),DataTypes.FIELD("age", DataTypes.INT()),DataTypes.FIELD("city", DataTypes.STRING())),Rows.rowOf("Alice",25, "Beijing"),Rows.rowOf("Bob",30, "Shanghai"),Rows.rowOf("Charlie",35, "Hangzhou")
)

(3)执行查询

在Table对象上可以执行各种查询操作,比如filter、select、group by等。例如,以下代码对Table进行了一个简单的select操作:

val result = table.select("name, age").where("age >30")

(4)输出结果

最后,可以通过toRetractStream()方法将Table对象转换为DataStream输出结果。例如,以下代码将查询结果输出到控制台:

result.toRetractStream[(String, Int)].print()

2.SQL的基本使用

(1)创建TableEnvironment对象:

和Table API一样,SQL也需要TableEnvironment对象来执行查询等操作。可以通过以下方式创建TableEnvironment对象:

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = StreamTableEnvironment.create(env)

(2)注册表:

在SQL中,需要将数据源注册为表,并为其定义表架构。可以通过以下代码将DataStream注册为表:

val ds: DataStream[(String, Int)] = ...
tEnv.createTemporaryView("my_table", ds, 'name, 'age)

其中,'name和'age是数据源中的字段名,用于定义表架构。

(3)执行查询:

在SQL中,可以通过executeSql()方法执行SQL查询。例如,以下代码查询了my_table表中年龄大于30的记录:

val result = tEnv.executeSql("SELECT name, age FROM my_table WHERE age >30")

(4)输出结果:

和Table API一样,最后可以通过toRetractStream()方法将查询结果转换为DataStream输出结果。例如,以下代码将查询结果输出到控制台:

result.toRetractStream[(String, Int)].print()

四、Flink支持的DataType分析

Flink支持Java和Scala中的大部分数据类型。
Java Tuple和Scala Case Class。
Java POJO:Java实体类。
Primitive Type:默认支持Java和Scala基本数据类型。
General Class Type:默认支持大多数Java和Scala Class。
Hadoop Writable:支持Hadoop中实现了org.apache.Hadoop.Writable的数据类型。
Special Type:比如Scala中的Either Option和Try。

根据类型分组:

1.基本数据类型:BOOLEAN、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE、DECIMAL、CHAR、VARCHAR、BINARY、VARBINARY。

2.时间类型:DATE、TIME、TIMESTAMP、INTERVAL YEAR、INTERVAL MONTH、INTERVAL DAY、INTERVAL HOUR、INTERVAL MINUTE、INTERVAL SECOND。

3.复合类型:ARRAY、MAP、ROW。

4.未知类型:NULL、RAW。

Flink通过DataType来描述表中列的数据类型,从而进行类型检查和类型推断。同时,Flink还支持使用UDF(User-Defined Function)自定义的数据类型。

相关内容

热门资讯

微软安卓系统叫什么,Windo... 你知道吗?在科技界,微软这个巨头最近可是搞了个大动作,竟然把目光投向了安卓系统!没错,就是那个我们日...
安卓系统没有最近任务,揭秘无最... 你是不是也遇到了这个问题?安卓系统里怎么就没有“最近任务”这个功能呢?别急,让我来给你详细说说这个事...
安卓系统怎么拒聊天,安卓系统聊... 你是不是也和我一样,有时候手机里聊天软件的消息太多,让人头都大了?别急,今天就来教你怎么在安卓系统里...
下载工资软件安卓系统,高效便捷... 你有没有想过,每个月的工资一到手,是不是就感觉整个人都轻松了呢?但是,你知道怎么轻松地管理你的工资吗...
下载灰灰影音安卓系统,畅享高清... 你有没有想过,一部手机,一部电脑,就能让你随时随地享受高清电影、热门剧集的乐趣?没错,这就是我们今天...
安卓系统是不是中国,中国智慧与... 你有没有想过,那个陪伴你每天刷手机、玩游戏、办公的安卓系统,它是不是中国的“孩子”呢?这个问题听起来...
电脑如果安装安卓系统,探索安卓... 你有没有想过,如果家里的电脑突然决定要换换口味,不再坚守Windows的阵营,而是拥抱安卓系统呢?想...
安卓手机升级苹果系统,体验全新... 你有没有想过,你的安卓手机突然间变成了苹果的忠实粉丝,想要体验一番iOS系统的魅力呢?这可不是天方夜...
安卓系统短信不通知,享受宁静 你是不是也遇到了这个问题?安卓系统短信不通知,是不是让你抓狂?别急,今天就来给你详细解析一下这个让人...
夏普电视非安卓系统,非安卓系统... 亲爱的读者们,你是否曾对夏普电视的非安卓系统感到好奇呢?今天,就让我带你一探究竟,揭开这个神秘面纱背...
安卓系统43适配软件,软件升级... 你有没有发现,你的安卓手机最近是不是有点儿“水土不服”?别急,别急,让我来给你揭秘为什么你的安卓系统...
安卓系统有车载系统吗吗,智能驾... 你有没有想过,当你坐在车里,享受着旅途的悠闲时光,手机上的安卓系统是不是也能派上用场呢?没错,我就要...
安卓8.1系统解锁方法,畅享自... 你有没有想过,你的安卓手机里隐藏着无数的小秘密?比如,解锁安卓8.1系统,就能让你的手机焕发出全新的...
安卓系统怎么打开邮件,安卓系统... 你有没有想过,每天那么多邮件,怎么才能快速打开它们呢?尤其是当你用的是安卓系统手机的时候。别急,今天...
封闭安卓系统安装软件,一步到位... 你有没有想过,为什么你的安卓手机里有些软件只能通过官方渠道安装呢?今天,就让我带你一探究竟,揭开封闭...
小米mipad升级安卓系统,解... 你有没有发现,最近小米Mipad的安卓系统升级可是个大热门呢!这不,我就迫不及待地来和你聊聊这个话题...
哪个安卓系统体验好,揭秘安卓系... 你有没有想过,手机里的安卓系统就像是个大厨,不同的版本就是不同的拿手好菜,有的让你回味无穷,有的却让...
安卓系统开发测试,全方位技术解... 你有没有想过,那个陪伴你每天刷手机、玩游戏、办公的安卓系统,是怎么从无到有,一步步成长起来的呢?今天...
安卓系统怎么查配置,轻松掌握设... 你有没有想过,你的安卓手机里藏着多少秘密?别小看这些配置信息,它们可是了解你手机性能的“小侦探”呢!...
pve下安装安卓系统,PVE环... 你有没有想过,在你的PVE服务器上安装一个安卓系统呢?听起来是不是有点酷炫?想象你的服务器不仅能够运...