Flink学习笔记窗口详解

Posted 2023-03-28 缘来如此09

一、概述：

        Flink中Batch是Streaming的一个特例，因此Flink底层引擎是一个流式引擎，在上面实现了流处理和批处理。而Window就是从Streaming到Batch的桥梁。

二、flink窗口类型：

Flink 提供了丰富的窗口类型，可以根据具体的场景和需求选择不同的窗口类型例如：

1.滚动窗口（Tumbling Windows）：将数据流按照固定大小的窗口进行划分，每个窗口内的数据处理完毕后，窗口的计算结果会被输出。

2.滑动窗口（Sliding Windows）：将数据流按照固定的大小和固定的滑动步长进行划分，每个窗口内的数据处理完毕后，窗口的计算结果会被输出。

3.会话窗口（Session Windows）：将数据流按照时间间隔进行分割，每个分段内的数据处理完毕后，窗口的计算结果会被输出。

4.全局窗口（Global Windows）：将整个数据流作为一个窗口进行处理，可以用来进行全局的计算操作。

5.增量窗口（Incremental Windows）：在数据流的处理过程中，持续对数据进行计算，而不是等待整个窗口的数据全部到齐后再进行计算。

6.Purging Window：清除窗口，当窗口中的数据达到一定的数量或时间时，将窗口中的数据全部清除。可以使用CountTrigger和PurgingTrigger结合Evictor函数来实现。

7.Evicting Window：驱逐窗口，当窗口中的数据达到一定的数量或时间时，将旧数据驱逐出窗口。可以使用CountTrigger和Evictor函数来实现。

8.Custom Trigger：自定义触发器，根据业务需求自定义触发器行为，可以使用Trigger接口来实现。

三、Window聚合分类：

Window聚合操作主要分为两种：一种是增量聚合，一种是全量聚合。

1.增量聚合：

增量聚合是指窗口每进入一条数据就计算一次

2. 全量聚合：

全量聚合是指在窗口触发的时候才会对窗口内的所有数据进行一次计算。

3.其他聚合方式：

 flink还有很多其他的聚合方式如：时间窗口聚合按照时间范围对数据进行分组聚合，计数窗口：按照数据条数对数据进行分组聚合等等，通过这些不同的聚合方式最终造就了flink的各式各样的窗口类型，也帮助使用者更加灵活的处理数据需求

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