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)自定义的数据类型。

相关内容

热门资讯

自动打开应用安卓系统,安卓系统... 你有没有想过,手机里的那些应用,有时候真是让人又爱又恨呢?有时候,我们急需某个应用,却得费老大力气去...
安卓系统防沉迷软件,守护青少年... 你有没有发现,现在手机上玩游戏的诱惑力简直让人无法抗拒?尤其是安卓系统,那丰富的游戏资源,简直让人停...
流量最快的安卓系统,揭秘流量最... 你有没有想过,为什么你的手机总是那么卡,而别人的手机却像开了挂一样流畅?是不是好奇,为什么有些安卓系...
小米5换换安卓系统,畅享极致性... 你有没有想过,你的小米5手机,那个陪伴你走过无数日夜的小家伙,是不是也该给它来个“换新装”了呢?没错...
国产的安卓系统手机,畅享智能生... 你有没有发现,最近国产的安卓系统手机越来越火了?没错,就是那种咱们自己研发的系统,那种让外国品牌都不...
安卓系统刷入停止,探究原因与解... 你有没有遇到过这种情况?手机刷机过程中突然停止了,安卓系统刷入停滞不前,心里那个急啊!别慌,今天就来...
汽车是安卓系统嘛,安卓系统在智... 你有没有想过,汽车里那个神奇的操作系统,是不是和安卓手机里的一样呢?没错,今天咱们就来聊聊这个话题—...
网易狼人杀 安卓系统,体验指尖... 亲爱的玩家们,你是否曾在深夜里,手机屏幕前,与一群好友展开一场惊心动魄的“狼人杀”对决?今天,就让我...
小米安卓系统小主机,探索小米安... 你有没有想过,家里的电视、电脑、平板,甚至手机,其实都可以变成一个超级智能的娱乐中心?没错,这就是小...
卡刷安卓系统大全,全面解析各类... 你有没有想过,你的安卓手机可以像变形金刚一样,随心所欲地变换模样?没错,今天就要给你揭秘一个神奇的世...
安卓系统测试流畅度,安卓系统流... 你有没有发现,现在手机更新换代的速度简直就像坐上了火箭呢!尤其是安卓系统,每次更新都让人眼前一亮。但...
安卓系统50怎么升级,轻松迈向... 亲爱的安卓用户们,你是否也像我一样,对安卓系统的更新充满了期待?没错,就是那个让我们的手机焕然一新的...
安卓5.1.1操作系统,系统特... 你知道吗?在手机世界里,操作系统就像是个大管家,它不仅决定了手机的脸面,还掌管着手机的所有“家务事”...
手机安卓系统如果升级,体验流畅... 亲爱的手机控们,你们有没有发现,你的安卓手机最近是不是总在提醒你更新系统呢?别急,别急,今天就来给你...
安卓系统怎么禁止待机,安卓系统... 手机待机时间短,是不是让你头疼不已?别急,今天就来教你一招,让你的安卓手机告别“短命”模式,延长待机...
亿联安卓苹果系统,跨平台沟通新... 你知道吗?在科技飞速发展的今天,手机操作系统可是咱们日常生活中不可或缺的一部分。说起手机系统,亿联安...
smoothx安卓系统安装ap... 你有没有想过,为什么你的手机里总是乱糟糟的,各种app堆在一起,找起来费劲得很?别急,今天就来教你怎...
安卓系统图库在哪里,图库应用位... 你有没有发现,手机里的照片越来越多,有时候想找一张特定的照片,却像大海捞针一样困难?别急,今天就来告...
安卓7.0系统自带彩蛋,隐藏彩... 你知道吗?安卓7.0系统里竟然藏着不少小秘密,就像一颗颗隐藏的彩蛋,等着我们去发现。今天,就让我带你...
安卓系统好用的电池,好用到飞起... 你有没有发现,用安卓手机的时候,电池续航能力简直让人爱不释手啊!没错,今天咱们就来聊聊这个话题——安...