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

相关内容

热门资讯

王者定位怎么关安卓系统,轻松实... 你是不是也和我一样,对王者荣耀这款游戏爱得深沉呢?不过,有时候游戏里的设置让人头疼,比如安卓系统的王...
树莓派安卓系统流畅,打造便携式... 亲爱的读者们,你是否曾想过,将树莓派与安卓系统结合,会擦出怎样的火花呢?今天,就让我带你一起探索这个...
安卓系统智能机顶盒,引领家庭娱... 你有没有想过,家里的电视也能变得智能起来?没错,就是那个陪伴我们多年的老电视,现在也能摇身一变,成为...
安卓系统很差了吗现在,性能优劣... 最近是不是有不少朋友在讨论安卓系统的问题呢?有人说它越来越差了,也有人觉得它还是那个熟悉的“老朋友”...
安卓系统uc安装包,Andro... 你有没有发现,手机里的安卓系统越来越强大了?今天,咱们就来聊聊这个话题——安卓系统中的UC安装包。你...
安卓系统谷歌能删吗,谷歌能否删... 你有没有想过,那个一直陪伴你手机生活的安卓系统,它背后的谷歌爸爸,是不是也能被你随意删掉呢?这可不是...
安卓系统会不会更耗电,解析其功... 你有没有发现,手机用着用着,电池就有点不给力了?尤其是那些用安卓系统的手机,有时候感觉电就像流水一样...
安卓系统中无效目录,安卓系统无... 你有没有遇到过在安卓系统中,明明文件夹就在那里,但是就是找不到的情况?别急,今天就来给你揭秘安卓系统...
国产安卓机哪个系统好用,探寻最... 你有没有想过,国产安卓机哪个系统最好用呢?这可是个让人纠结的问题,毕竟每个系统都有它的特色和亮点。今...
安卓系统cpua9,引领性能与... 你有没有发现,最近你的安卓手机运行得是不是比以前顺畅多了?这可多亏了那个强大的安卓系统CPUA9啊!...
安卓系统usb驱动程序,功能、... 你有没有遇到过这种情况:手机里存了那么多宝贝照片和视频,想传输到电脑上保存,结果电脑却像个小顽皮,死...
安卓操作系统怎么关闭,轻松关闭... 手机里的安卓操作系统是不是有时候让你觉得有点儿烦呢?别急,今天就来手把手教你如何轻松关闭安卓操作系统...
追星手机壳推荐安卓系统,盘点热... 你有没有发现,现在追星族们对手机壳的热爱简直到了疯狂的地步?没错,就是那种能让你一秒变身偶像迷妹的手...
ios系统用安卓系统游戏下载软... 你有没有想过,明明是iOS系统的手机,却想玩安卓系统的游戏?这可不是什么天方夜谭,现在就有这么神奇的...
安卓高系统怎么用美化,打造专属... 亲爱的安卓用户们,你是不是也和我一样,对手机系统美化情有独钟呢?想要让你的安卓手机焕然一新,变得个性...
安卓系统怎么开夜间模式,安卓系... 亲爱的手机控们,你是不是在夜晚使用安卓手机时,眼睛感到有些不适?别担心,今天我要给你揭秘一个超级实用...
王者安卓系统用苹果人脸,一场视... 你知道吗?最近在手机圈里可是掀起了一股不小的波澜呢!那就是王者安卓系统竟然用上了苹果人脸识别技术!是...
安卓444怎么升级系统,轻松迈... 你那安卓444的小家伙是不是已经有点儿落伍了?别急,今天就来给你详细说说怎么给它来个系统升级,让它焕...
安卓系统raw修图软件,探索安... 你有没有发现,手机拍照越来越方便了,但有时候拍出来的照片还是不够完美呢?别急,今天就来给你安利几款安...
安卓系统的王者切换苹果,从安卓... 你知道吗?最近身边的朋友圈里掀起了一股热潮,那就是安卓系统的王者们纷纷切换到苹果阵营。这可真是让人大...