4.2.8 Flink-流处理框架-Table API 与 SQL-流处理中的特殊概念之时间特性

古城微笑少年丶 2022-11-09 00:58 151阅读 0赞

**目录**

1.写在前面

2.时间特性

2.1 定义处理时间（Processing Time）

2.1.1 DataStream 转化成 Table 时指定

2.1.2 定义 Table Schema 时指定

2.1.3 创建表的 DDL 中指定

2.2 定义事件时间（Event Time）

2.2.1 DataStream 转化成 Table 时指定

2.2.2 定义 Table Schema 时指定

2.2.3 在创建表的 DDL 中定义

--------------------

# 1.写在前面 #

基于时间的操作（比如 Table API 和 SQL 中窗口操作），需要定义相关的时间语义和时间数据来源的信息。所以，Table 可以提供一个逻辑上的时间字段，用于在表处理程序中，指 示时间和访问相应的时间戳。

时间属性，可以是每个表 schema 的一部分。一旦定义了时间属性，它就可以作为一个 字段引用，并且可以在基于时间的操作中使用。 **时间属性的行为类似于常规时间戳，可以访问，并且进行计算**。

# 2.时间特性 #

## 2.1 定义处理时间（Processing Time） ##

处理时间语义下，允许表处理程序根据机器的本地时间生成结果。它是时间的最简单概念。它既不需要提取时间戳，也不需要生成 watermark。 定义处理时间属性有三种方法：**在 DataStream 转化时直接指定**；**在定义 Table Schema 时指定**；**在创建表的 DDL 中指定**。

### 2.1.1 DataStream 转化成 Table 时指定 ###

由 DataStream 转换成表时，可以在后面指定字段名来定义 Schema。在定义 Schema 期间，可以使用.proctime，定义处理时间字段。 注意，这个 proctime 属性只能通过附加逻辑字段，来扩展物理 schema。因此，只能在 schema 定义的末尾定义它。

Table dataTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts, temperature as temp, pt.proctime");

上面代码系统会自动的在我们的表中添加一个字段，将我们的系统时间添加上去。精确到毫秒的一个tampstamp。

![20210314122131200.png][]

### 2.1.2 定义 Table Schema 时指定 ###

这种方法其实也很简单，只要在定义 Schema 的时候，加上一个新的字段，并指定成 proctime 就可以了。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

### 2.1.3 创建表的 DDL 中指定 ###

在创建表的 DDL 中，增加一个字段并指定成 proctime，也可以指定当前的时间字段。注意：**运行这段 DDL，必须使用 Blink Planner**。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

## 2.2 定义事件时间（Event Time） ##

事件时间语义，允许表处理程序根据每个记录中包含的时间生成结果。这样即使在有乱 序事件或者延迟事件时，也可以获得正确的结果。 为了处理无序事件，并区分流中的准时和迟到事件；Flink 需要从事件数据中，提取时间戳，并用来推进事件时间的进展（watermark）。

定义事件时间属性有三种方法：**在 DataStream 转化时直接指定**；**在定义 Table Schema 时指定**；**在创建表的 DDL 中指定**。

### 2.2.1 DataStream 转化成 Table 时指定 ###

在将数据流转换为表时，有两种定义时间属性的方法。根据指定的.rowtime 字段名是否 存在于数据流的架构中，timestamp 字段可以： ⚫ 作为新字段追加到 schema ⚫ 替换现有字段。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

### 2.2.2 定义 Table Schema 时指定 ###

这种方法只要在定义 Schema 的时候，将事件时间字段，并指定成 rowtime 就可以了。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3][]

### 2.2.3 在创建表的 DDL 中定义 ###

事件时间属性，是使用 CREATE TABLE DDL 中的 WARDMARK 语句定义的。watermark 语句，定义现有事件时间字段上的 watermark 生成表达式，该表达式将事件时间字段标记为事件时间属性。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4][]

这里 FROM\_UNIXTIME 是系统内置的时间函数，用来将一个整数（秒数）转换成 “YYYY-MM-DD hh:mm:ss”格式（默认，也可以作为第二个 String 参数传入）的日期时间字符串（date time string）；然后再用 TO\_TIMESTAMP 将其转换成 Timestamp。

[20210314122131200.png]: /images/20221023/73cea9c0053f44a1824ac90ecd8df25f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/9345f5cf8c0c46caa7dc707b5521ea98.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/9004234fc80446afb3e6ac2ff9a5d8d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221023/03d2096e0e8a47878b3dc63967bac987.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 3]: /images/20221023/80f49f261a4343b2b9dacf6d4f956972.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 4]: /images/20221023/2947680edd1640fb956a7e07f3a5734d.png