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

相关内容

热门资讯

安卓系统的第二套系统,创新与变... 你知道吗?在科技飞速发展的今天,安卓系统已经成为了智能手机市场上的霸主。但是,你知道吗?安卓系统其实...
全军出击安卓系统版本,战力再攀... 你有没有发现,最近全军出击这款游戏在安卓系统上的版本更新可是越来越频繁了呢?这不,我就来给你好好扒一...
安卓系统热点限速软件,优化热点... 你有没有遇到过这种情况:手机连接热点后,网速就像蜗牛爬行一样慢,简直让人抓狂!别急,今天就来给你揭秘...
安卓系统占内存多,揭秘内存消耗... 你有没有发现,手机用着用着,内存就不够用了?尤其是安卓系统,好像特别能吃内存,让人头疼不已。今天,就...
最近安卓系统奔溃,揭秘原因与应... 最近手机界可是炸开了锅呢!安卓系统竟然出现了大规模奔溃,这可真是让人摸不着头脑。咱们一起来探究这背后...
ce系统能刷安卓系统吗,揭秘能... 你有没有想过,你的安卓手机是不是也能用上CE系统呢?这可不是天方夜谭,今天就来给你揭秘一下这个神秘的...
安卓系统UI设计特色,创新与用... 你有没有发现,每次打开安卓手机,那界面设计得真是让人眼前一亮呢?今天,就让我带你一起探索一下安卓系统...
ipod有安卓系统吗,跨界融合... 你有没有想过,那个曾经风靡一时的iPod,它到底有没有安卓系统呢?这个问题,估计让不少音乐爱好者都好...
安卓多少系统最高的,揭秘最高版... 你有没有想过,你的安卓手机到底升级到了哪个系统版本呢?是不是好奇安卓系统里哪个版本才是最高级的呢?别...
现在安卓最高的系统,揭秘安卓1... 你有没有发现,手机更新换代的速度简直就像坐上了火箭呢!这不,最近安卓系统又来了一次大升级,听说这是现...
安卓系统怎么隐藏相册,安卓系统... 你是不是也有那么几本“私人珍藏”,不想让旁人随意翻看呢?比如,手机里的相册,里面藏着我们的喜怒哀乐,...
安卓桌面挂件系统下载,下载与个... 你有没有发现,手机桌面上的那些小玩意儿,简直就是生活的调味品?今天,咱们就来聊聊安卓桌面挂件系统下载...
wp手机加安卓系统,探索跨界新... 你有没有想过,为什么你的手机总是那么卡,而别人的手机却流畅得像风一样?是不是觉得自己的手机有点OUT...
省电手机推荐安卓系统,安卓系统... 手机这玩意儿,对于我们这些手机控来说,简直就是生活的必需品。但是,你知道吗?现在市面上那么多手机,要...
安卓系统衰落怎么恢复,探寻衰落... 你有没有发现,最近安卓系统好像有点儿“水土不服”了呢?曾经的霸主地位,如今似乎有些动摇。不过别急,今...
安卓系统手机应用锁,安全无忧的... 你有没有发现,现在手机里的秘密越来越多,是不是也跟小秘密一样,想要找个地方藏起来呢?没错,今天就要来...
安卓系统书院源app,安卓系统... 你有没有发现,手机里的安卓系统越来越智能了?今天,我要给你介绍一个特别有意思的书院源app,它可是安...
安卓系统8.1平板推荐,安卓8... 你有没有想过,拥有一款性能卓越、体验流畅的安卓系统8.1平板,简直就是移动办公和娱乐的完美搭档?没错...
谷歌不给华为安卓系统,探索替代... 你知道吗?最近科技圈可是炸开了锅!谷歌突然宣布,不给华为提供安卓系统了!这可不仅仅是两家公司之间的小...
选择安卓系统原因调查,揭秘安卓... 你有没有想过,为什么那么多人会选择安卓系统呢?是不是好奇他们到底看中了安卓的哪些“小秘密”?今天,就...