一、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的区别：
-
数据类型：DataSet API适用于处理有界数据，即离线批处理；DataStream API适用于处理无界数据，即实时流处理。
-
数据处理方式：DataSet API采用批处理方式，即将数据读取到内存中，进行批量计算，然后将结果写出；DataStream API采用流处理方式，即从数据流中逐个读取数据，进行实时计算，然后将结果发送到下游。
-
时间处理：在DataStream API中，时间处理非常重要，可以通过时间窗口、时间滑动窗口等方式对数据进行处理。而在DataSet API中，时间处理相对简单，通常只需要使用时间戳进行排序和分组即可。
-
窗口：DataStream API中支持各种窗口，如滚动窗口、滑动窗口、会话窗口等，而DataSet API中则不支持窗口。
-
稳定性：由于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。文章来源：https://www.toymoban.com/news/detail-402622.html
Flink通过DataType来描述表中列的数据类型，从而进行类型检查和类型推断。同时，Flink还支持使用UDF（User-Defined Function）自定义的数据类型。文章来源地址https://www.toymoban.com/news/detail-402622.html
到了这里，关于Flink学习笔记（二）Flink常用API详解的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
