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

相关内容

热门资讯

编程安卓系统和鸿蒙主题,跨平台... 你有没有想过,手机的世界里,除了苹果的iOS和安卓的操作系统,还有个神秘的鸿蒙系统?今天,咱们就来聊...
哪个安卓机系统好用,探索安卓系... 你有没有想过,手机里的安卓系统就像是个大厨,不同的系统就像不同的烹饪手法,有的让你吃得津津有味,有的...
安卓如何控制苹果系统,从安卓到... 你知道吗?在这个科技飞速发展的时代,安卓和苹果两大操作系统之间的较量从未停歇。虽然它们各自有着忠实的...
安卓原生系统文件夹,安卓原生系... 你有没有发现,每次打开安卓手机,里面那些文件夹就像是一个个神秘的宝箱,里面藏着各种各样的宝贝?今天,...
基于安卓系统的游戏开发,从入门... 你有没有想过,为什么安卓手机上的游戏总是那么吸引人?是不是因为它们就像是你身边的好朋友,随时随地都能...
安卓系统怎样装驱动精灵,安卓系... 你那安卓设备是不是突然间有点儿不给力了?别急,今天就来手把手教你如何给安卓系统装上驱动精灵,让你的设...
如何本地安装安卓系统包,详细步... 你有没有想过,把安卓系统装在你的电脑上,是不是就像给电脑穿上了时尚的新衣?想象你可以在电脑上直接玩手...
安卓12卡刷系统教程,体验全新... 你有没有发现,你的安卓手机最近有点儿不给力了?运行速度慢得像蜗牛,是不是也想给它来个“换血大法”,让...
安卓系统无法打开swf文件,安... 最近是不是发现你的安卓手机有点儿不给力?打开SWF文件时,是不是总是出现“无法打开”的尴尬局面?别急...
鸿蒙系统依赖于安卓系统吗,独立... 你有没有想过,我们手机里的那个鸿蒙系统,它是不是真的完全独立于安卓系统呢?这个问题,估计不少手机控都...
适合安卓系统的图片软件,精选图... 手机里堆满了各种美美的照片,是不是觉得找起来有点头疼呢?别急,今天就来给你安利几款超级适合安卓系统的...
阴阳师安卓系统典藏,探寻阴阳师... 亲爱的阴阳师们,你是否在安卓系统上玩得如痴如醉,对那些精美的典藏式神们垂涎欲滴?今天,就让我带你深入...
安卓系统有碎片化缺点,系统优化... 你知道吗?在手机江湖里,安卓系统可是个响当当的大侠。它那开放、自由的个性,让无数手机厂商和开发者都为...
安卓4系统手机微信,功能解析与... 你有没有发现,现在市面上还有很多安卓4系统的手机在使用呢?尤其是那些喜欢微信的朋友们,这款手机简直就...
鸿蒙系统是安卓的盗版,从安卓“... 你知道吗?最近在科技圈里,关于鸿蒙系统的讨论可是热闹非凡呢!有人说是安卓的盗版,有人则认为这是华为的...
安卓系统怎么剪辑音乐,轻松打造... 你是不是也和我一样,手机里存了超多好听的歌,但是有时候想给它们来个变身,变成一段专属的旋律呢?别急,...
怎么把安卓手机系统变为pc系统... 你有没有想过,把你的安卓手机变成一台PC呢?听起来是不是有点酷炫?想象你可以在手机上玩电脑游戏,或者...
手机怎么装安卓11系统,手机安... 你有没有想过,让你的手机也来个“青春焕发”,升级一下系统呢?没错,就是安卓11系统!这个新系统不仅带...
安卓系统如何拼网络,构建高效连... 你有没有想过,你的安卓手机是怎么和网络“谈恋爱”的呢?没错,就是拼网络!今天,就让我带你一探究竟,看...
安卓系统怎么看小说,轻松畅享电... 你有没有发现,手机里装了那么多应用,最离不开的竟然是那个小小的小说阅读器?没错,就是安卓系统上的小说...