大数据Flink（一百一十六）：Flink SQL的时间属性

文章目录
Flink SQL的时间属性
一、Flink 三种时间属性简介
二、Flink 三种时间属性的应用场景
三、​​​​​​​SQL 指定时间属性的两种方式
四、​​​​​​​​​​​​​​SQL 处理时间DDL界说
五、​​​​​​​​​​​​​​SQL 事件时间DDL界说

---

Flink SQL的时间属性

先看一下本文团体的思路：

• 与离线处理中常见的时间分区字段一样，在实时处理中，时间属性也是一个核心概念。Flink 支持 处理时间、事件时间、摄入时间 三种时间语义。
  
• 分别介绍三种时间语义的应用场景及案例。三种时间在生产环境的利用频次事件时间（SQL 常用） > 处理时间（SQL 几乎不用，DataStream 少用） > 摄入时间（不用）
  

 
一、Flink 三种时间属性简介

• 事件时间：指的是数据自己携带的时间，这个时间是在事件产生时的时间，而且在 Flink SQL 触发盘算时，也利用数据自己携带的时间。这就叫做 事件时间。目宿世产环境中用的最多。
  
• 处理时间：指的是具体算子盘算数据实验时的机器时间（例如在算子中 Java 取 System.currentTimeMillis()) ），在生产环境中用的次多。
  
• 摄入时间：指的是数据从数据源进入 Flink 的时间。摄入时间用的最少，可以说根本倒霉用。
  

要注意到：

• 上述的三种时间概念不是由于有了数据而诞生的，而是有了 Flink 之后根据现实的应用场景而诞生的。以事件时间举个例子，假如只是数据携带了时间，Flink 也消耗了这个数据，但是在 Flink 中没有利用数据的这个时间作为盘算的触发条件，也不能把这个 Flink 使命叫做事件时间的使命。
  
• 其次，要认识到，一般一个 Flink 使命只会有一个时间属性，以是时间属性通常认为是一个使命粒度的。举例：我们可以说 A 使命是事件时间语义的使命，B 使命是处理时间语义的使命。固然了，一个使命也可以存在多个时间属性。
  

二、Flink 三种时间属性的应用场景

以上三种时间属性到底对我们的使命有啥影响呢？三种时间属性的应用场景是啥？
先说结论，在 Flink 中时间的作用：

• 重要表现在包含时间窗口的盘算中：用于标识使命的时间进度，来判断是否需要触发窗口的盘算。好比常用的滚动窗口、滑动窗口等都需要时间推动触发。这些窗口的应用场景后续会详细介绍。
  
• 次要表现在自界说时间语义的盘算中：举个例子，好比用户可以自界说每隔 10s 的本地时间，或者消耗到的数据的时间戳每增大 10s，就把盘算结果输出一次，时间在此类应用中也是一种标识使命进度的作用。
  

以 滚动窗口 的聚合使命为例来介绍一下事件时间和处理时间的对比区别。

• 事件时间案例：还是以之前的 clicks 表拿来举例。
  

上面这个案例的窗口大小是 1 小时，需求方需要按照用户点击时间戳 cTime 划分数据（划分滚动窗口），然后盘算出 count 聚合结果（如许盘算能反映出事件的真实发生时间），那么就需要把 cTime 设置为窗口的划分时间戳，即代码中 tumble(cTime, interval '1' hour)。
上面这种就叫做事件时间。即用数据中自带的时间戳进行窗口的划分（点击操作真实的发生时间）。
后续 Flink SQL 使命在运行的过程中也会现实按照 cTime 的当前时间作为一小时窗口结束触发条件并盘算一个小时窗口内的数据。

• 处理时间案例：还是以之前的 clicks 表拿来举例。
  

还是上面谁人案例，但是这次需求方不需要按照数据上的时间戳划分数据（划分滚动窗口），只需要数据来了之后， 在 Flink 机器上的时间作为一小时窗口结束的触发条件并盘算。
那么这种触发机制就是处理时间。

• 摄入时间案例：在 Flink 从外部数据源读取到数据时，给这条数据带上的当前数据源算子的本地时间戳。卑鄙可以用这个时间戳进行窗口聚合，不过这种几乎倒霉用。
  

 
三、​​​​​​​​​​​​​​SQL 指定时间属性的两种方式

假如要满足 Flink SQL 时间窗口类的聚合操作，SQL 或 Table API 中的 数据源表 就需要提供时间属性（相当于我们把这个时间属性在 数据源表 上面进行声明），以及支持时间相关的操作。
那么来看看 Flink SQL 为我们提供的两种指定时间戳的方式：

• CREATE TABLE DDL 创建表的时间指定（保举）
  
• 可以在 DataStream 中指定，在后续的 DataStream 转的 Table 中利用（略过，讲课以Flink SQL为主）
  

一旦时间属性界说好，它就可以像普通列一样利用，也可以在时间相关的操作中利用。
 
四、​​​​​​​​​​​​​​SQL 处理时间DDL界说

处理时间语义下，利用当前机器的体系时间作为处理时间。它是时间的最简朴概念。它既不需要提取时间戳，也不需要生成watermark。
来看看 Flink SQL 中怎样指定处理时间。

• CREATE TABLE DDL 指定时间戳的方式。
  

1. CREATE TABLE user_actions (
2.   user_name STRING,
3.   data STRING,
4.   -- 使用下面这句来将 user_action_time 声明为处理时间
5.   user_action_time AS PROCTIME()
6. ) WITH (
7.   ...
8. );

复制代码

• 利用案例
  

点击Flink开辟平台左侧资源管理，点击上传资源，将资料中的order.csv文件进行上传。

可以点击复制按钮复制其在oss的路径。

可以在oss对应路径看到此文件

读取order.csv'文件的数据，在原本的Schema上添加一个假造的时间戳列，时间戳列由PROCTIME()函数盘算产生。建表语句如下（path背面路径需要修改） 

1. create table proctime_ddl_table (
2. `userid` varchar,
3. `timestamp` bigint,
4. `money` double,
5. `category` varchar,
6. `pt` AS PROCTIME()
7. ) with (
8. 'connector' = 'filesystem',
9. 'path' = 'oss://lanson-bucket/artifacts/namespaces/lanson-workspace-default/order.csv',
10. 'format' = 'csv'
11. );

复制代码

查询表数据（调试） 

1. select * from proctime_ddl_table;

复制代码

查询结果如下

点击左侧导航栏元数据管理，查看表信息。 

五、​​​​​​​​​​​​​​SQL 事件时间DDL界说

来看看 Flink 中怎样指定事件时间。
Event Time时间语义利用一条数据现实发生的时间作为时间属性，在Table API & SQL中这个字段通常被称为rowtime。这种模式下多次重复盘算时，盘算结果是确定的。这意味着，Event Time时间语义可以保证流处理和批处理的同一。
Event Time时间语义下，我们需要设置每条数据发生时的时间戳，并提供一个Watermark。Watermark表示迟于该时间的数据都作为迟到数据对待。

• CREATE TABLE DDL 指定时间戳的方式。
  

1. CREATE TABLE user_actions (
2.   user_name STRING,
3.   data STRING,
4.   user_action_time TIMESTAMP(3),
5.   -- 使用下面这句来将 user_action_time 声明为事件时间，并且声明 watermark 的生成规则，即 user_action_time 减 5 秒
6.   -- 事件时间列的字段类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型
7.   WATERMARK FOR user_action_time AS user_action_time - INTERVAL '5' SECOND
8. ) WITH (
9.   ...
10. );

复制代码

在上面的DDL中，WATERMARK起到了界说Event Time时间属性的作用，在这里临时不教学，watermark知识点后续会讲到。

假如想利用事件时间，那么我们的时间戳类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型。
但是现实应用中时间戳一般都是秒或者是毫秒（BIGINT 类型），那这种环境怎么办？
解决方案如下：

1. CREATE TABLE user_actions (
2.   user_name STRING,
3.   data STRING,
4.   -- 1. 这个 ts 就是常见的毫秒级别时间戳
5.   ts BIGINT,
6.   -- 2. 将毫秒时间戳转换成 TIMESTAMP_LTZ 类型
7.   time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
8.   -- 3. 使用下面这句来将 time_ltz 声明为事件时间，并且声明 watermark 的生成规则，即 time_ltz 减 5 秒
9.   -- 事件时间列的字段类型必须是 TIMESTAMP 或者 TIMESTAMP_LTZ 类型
10.   WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' SECOND
11. ) WITH (
12.   ...
13. );

复制代码

• 利用案例
  

读取order.csv'文件的数据，界说现有事件时间字段上的 watermark 生成表达式，该表达式将事件时间字段标记为事件时间属性。
建表语句如下

1. create table eventime_ddl_table (
2. `userid` varchar,
3. `timestamp` bigint,
4. `money` double,
5. `category` varchar,
6. rt AS TO_TIMESTAMP(FROM_UNIXTIME(`timestamp`)),
7. watermark for rt as rt - interval '1' second
8. ) with (
9. 'connector' = 'filesystem',
10. 'path' = 'oss://bucketnanfeng/artifacts/namespaces/lanson-flinkworkspace1-default/order.csv',
11. 'format' = 'csv'
12. );

复制代码

查询表数据（调试）

1. select * from eventime_ddl_table;

复制代码

查询结果如下

点击左侧导航栏元数据管理，查看表信息。 

 

---