Flink学习笔记（二）Flink常用API详解

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