0
0

【Flink SQL】Flink SQL 基础概念（五）：SQL 时区问题

《Flink SQL 基础概念》系列，共包含以下 5 篇文章：

  • Flink SQL 基础概念（一）：SQL & Table 运行环境、基本概念及常用 API
  • Flink SQL 基础概念（二）：数据类型
  • Flink SQL 基础概念（三）：SQL 动态表 & 连续查询
  • Flink SQL 基础概念（四）：SQL 的时间属性
  • Flink SQL 基础概念（五）：SQL 时区问题

😊 如果您觉得这篇文章有用 ✔️ 的话，请给博主一个一键三连 🚀🚀🚀 吧 （点赞 🧡、关注 💛、收藏 💚）！！！您的支持 💖💖💖 将激励 🔥 博主输出更多优质内容！！！

Flink SQL 基础概念（五）：SQL 时区问题

1.SQL 时区解决的问题

首先说一下这个问题的背景：大家想一下离线 Hive 环境中，有遇到过时区相关的问题吗？

至少博主目前没有碰到过，因为这个问题在底层的数据集成系统都已经给解决了，小伙伴萌拿到手的 ODS 层表都是已经按照所在地区的时区给格式化好的了。

举个例子：小伙伴萌看到日期分区为 

2022-01-01

的 Hive 表时，可以默认认为该分区中的数据就对应到你所在地区的时区的 

2022-01-01

日的数据。

但是 Flink 中时区问题要特别引起关注，不加小心就会误用。而本节 SQL 时区旨在帮助大家了解到以下两个场景的问题：

  • 在 1.13 1.13 1.13 之前，DDL Create Table 中使用 PROCTIME() 指定处理时间列时，返回值类型为 TIMESTAMP(3) 类型，而 TIMESTAMP(3) 是不带任何时区信息的，默认为 UTC 时间（ 0 0 0 时区）。
  • 使用 StreamTableEnvironment::createTemporaryView 将 DataStream 转为 Table 时，注册处理时间（proctime.proctime）、事件时间列（rowtime.rowtime）时，两列时间类型也为 TIMESTAMP(3) 类型，不带时区信息。

而以上两个场景就会导致：

  • 在北京时区的用户使用 TIMESTAMP(3) 类型的时间列开最常用的 1 1 1 天的窗口时，划分出来的窗口范围是北京时间的 [2022-01-01 08:00:00, 2022-01-02 08:00:00]，而不是北京时间的 [2022-01-01 00:00:00, 2022-01-02 00:00:00]。因为 TIMESTAMP(3) 是默认的 UTC 时间，即 0 0 0 时区。
  • 北京时区的用户将 TIMESTAMP(3) 类型时间属性列转为 STRING 类型的数据展示时，也是 UTC 时区的，而不是北京时间的。

因此充分了解本节的知识内容可以很好的帮你避免时区问题错误。

2.SQL 时间类型

Flink SQL 支持 TIMESTAMP（不带时区信息的时间）、TIMESTAMP_LTZ（带时区信息的时间）

  • TIMESTAMP（不带时区信息的时间）：是通过一个 年， 月， 日， 小时， 分钟，秒 和 小数秒 的字符串来指定。举例：1970-01-01 00:00:04.001。 - 为什么要使用字符串来指定呢？因为此种类型不带时区信息，所以直接用一个字符串指定就好了。- 那 TIMESTAMP 字符串的时间代表的是什么时区的时间呢？UTC 时区，也就是默认 0 0 0 时区，对应中国北京是东八区。
  • TIMESTAMP_LTZ（带时区信息的时间）：没有字符串来指定，而是通过 Java 标准 Epoch 时间 1970-01-01T00:00:00Z 开始计算的毫秒数。举例：1640966400000。 - 其时区信息是怎么指定的呢？是通过本次任务中的时区配置参数 table.local-time-zone 设置的。- 时间戳本身也不带有时区信息，为什么要使用时间戳来指定呢？就是因为时间戳不带有时区信息，所以我们通过配置 table.local-time-zone 时区参数之后，就能将一个不带有时区信息的时间戳转换为带有时区信息的字符串了。举例：table.local-time-zone 为 Asia/Shanghai 时，4001 时间戳转化为字符串的效果是 1970-01-01 08:00:04.001。

3.时区参数生效的 SQL 时间函数

以下 SQL 中的时间函数都会受到时区参数的影响，从而做到最后显示给用户的时间、窗口的划分都按照用户设置时区之内的时间。

  • ✅ LOCALTIME
  • ✅ LOCALTIMESTAMP
  • ✅ CURRENT_DATE
  • ✅ CURRENT_TIME
  • ✅ CURRENT_TIMESTAMP
  • ✅ CURRENT_ROW_TIMESTAMP()
  • ✅ NOW()
  • ✅ PROCTIME()：其中 PROCTIME() 在 1.13 1.13 1.13 版本及之后版本，返回值类型是 TIMESTAMP_LTZ(3)

在 Flink SQL Client 中执行结果如下：

Flink SQL>SETsql-client.execution.result-mode=tableau;
Flink SQL>CREATEVIEW MyView1 ASSELECT LOCALTIME, LOCALTIMESTAMP,CURRENT_DATE,CURRENT_TIME,CURRENT_TIMESTAMP, CURRENT_ROW_TIMESTAMP(),NOW(), PROCTIME();
Flink SQL>DESC MyView1;+------------------------+-----------------------------+-------+-----+--------+-----------+|                   name |type|null|key| extras | watermark |+------------------------+-----------------------------+-------+-----+--------+-----------+|              LOCALTIME |TIME(0)|false|||||         LOCALTIMESTAMP |TIMESTAMP(3)|false|||||CURRENT_DATE|DATE|false|||||CURRENT_TIME|TIME(0)|false|||||CURRENT_TIMESTAMP|            TIMESTAMP_LTZ(3)|false|||||CURRENT_ROW_TIMESTAMP()|            TIMESTAMP_LTZ(3)|false|||||NOW()|            TIMESTAMP_LTZ(3)|false|||||             PROCTIME()| TIMESTAMP_LTZ(3)*PROCTIME*|false||||+------------------------+-----------------------------+-------+-----+--------+-----------+

Flink SQL>SETtable.local-time-zone=UTC;
Flink SQL>SELECT*FROM MyView1;+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+| LOCALTIME |          LOCALTIMESTAMP |CURRENT_DATE|CURRENT_TIME|CURRENT_TIMESTAMP| CURRENT_ROW_TIMESTAMP()|NOW()|              PROCTIME()|+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+|15:18:36|2021-04-1515:18:36.384|2021-04-15|15:18:36|2021-04-1515:18:36.384|2021-04-1515:18:36.384|2021-04-1515:18:36.384|2021-04-1515:18:36.384|+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+

Flink SQL>SETtable.local-time-zone=Asia/Shanghai;
Flink SQL>SELECT*FROM MyView1;+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+| LOCALTIME |          LOCALTIMESTAMP |CURRENT_DATE|CURRENT_TIME|CURRENT_TIMESTAMP| CURRENT_ROW_TIMESTAMP()|NOW()|              PROCTIME()|+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+|23:18:36|2021-04-1523:18:36.384|2021-04-15|23:18:36|2021-04-1523:18:36.384|2021-04-1523:18:36.384|2021-04-1523:18:36.384|2021-04-1523:18:36.384|+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+

Flink SQL>CREATEVIEW MyView2 ASSELECT TO_TIMESTAMP_LTZ(4001,3)AS ltz,TIMESTAMP'1970-01-01 00:00:01.001'AS ntz;
Flink SQL>DESC MyView2;+------+------------------+-------+-----+--------+-----------+| name |type|null|key| extras | watermark |+------+------------------+-------+-----+--------+-----------+|  ltz | TIMESTAMP_LTZ(3)|true|||||  ntz |TIMESTAMP(3)|false||||+------+------------------+-------+-----+--------+-----------+

Flink SQL>SETtable.local-time-zone=UTC;
Flink SQL>SELECT*FROM MyView2;+-------------------------+-------------------------+|                     ltz |                     ntz |+-------------------------+-------------------------+|1970-01-0100:00:04.001|1970-01-0100:00:01.001|+-------------------------+-------------------------+

Flink SQL>SETtable.local-time-zone=Asia/Shanghai;
Flink SQL>SELECT*FROM MyView2;+-------------------------+-------------------------+|                     ltz |                     ntz |+-------------------------+-------------------------+|1970-01-0108:00:04.001|1970-01-0100:00:01.001|+-------------------------+-------------------------+

Flink SQL>CREATEVIEW MyView3 ASSELECT ltz, CAST(ltz ASTIMESTAMP(3)), CAST(ltz AS STRING), ntz, CAST(ntz AS TIMESTAMP_LTZ(3))FROM MyView2;
Flink SQL>DESC MyView3;+-------------------------------+------------------+-------+-----+--------+-----------+|                          name |type|null|key| extras | watermark |+-------------------------------+------------------+-------+-----+--------+-----------+|                           ltz | TIMESTAMP_LTZ(3)|true|||||     CAST(ltz ASTIMESTAMP(3))|TIMESTAMP(3)|true|||||           CAST(ltz AS STRING)|           STRING |true|||||                           ntz |TIMESTAMP(3)|false||||| CAST(ntz AS TIMESTAMP_LTZ(3))| TIMESTAMP_LTZ(3)|false||||+-------------------------------+------------------+-------+-----+--------+-----------+

Flink SQL>SELECT*FROM MyView3;+-------------------------+---------------------------+-------------------------+-------------------------+-------------------------------+|                     ltz | CAST(ltz ASTIMESTAMP(3))|     CAST(ltz AS STRING)|                     ntz | CAST(ntz AS TIMESTAMP_LTZ(3))|+-------------------------+---------------------------+-------------------------+-------------------------+-------------------------------+|1970-01-0108:00:04.001|1970-01-0108:00:04.001|1970-01-0108:00:04.001|1970-01-0100:00:01.001|1970-01-0100:00:01.001|+-------------------------+---------------------------+-------------------------+-------------------------+-------------------------------+

4.事件时间和时区应用案例

这里分两类，分别是 TIMESTAMP（不带时区信息的时间）、TIMESTAMP_LTZ（带时区信息的时间）的事件时间 Flink SQL 任务

  • TIMESTAMP（不带时区信息的时间）
Flink SQL>CREATETABLE MyTable2 (
                  item STRING,
                  price DOUBLE,
                  ts TIMESTAMP(3),-- TIMESTAMP 类型的时间戳
                  WATERMARK FOR ts AS ts -INTERVAL'10'SECOND)WITH('connector'='socket','hostname'='127.0.0.1','port'='9999','format'='csv');

Flink SQL>CREATEVIEW MyView4 ASSELECT
                TUMBLE_START(ts,INTERVAL'10' MINUTES)AS window_start,
                TUMBLE_END(ts,INTERVAL'10' MINUTES)AS window_end,
                TUMBLE_ROWTIME(ts,INTERVAL'10' MINUTES)as window_rowtime,
                item,MAX(price)as max_price
            FROM MyTable2
                GROUPBY TUMBLE(ts,INTERVAL'10' MINUTES), item;

Flink SQL>DESC MyView4;+----------------+------------------------+------+-----+--------+-----------+|           name |type|null|key| extras | watermark |+----------------+------------------------+------+-----+--------+-----------+|   window_start |TIMESTAMP(3)|true|||||     window_end |TIMESTAMP(3)|true||||| window_rowtime |TIMESTAMP(3)*ROWTIME*|true|||||           item |                 STRING |true|||||      max_price |DOUBLE|true||||+----------------+------------------------+------+-----+--------+-----------+

将数据写入到 MyTable2 中：

> nc -lk 9999
A,1.1,2021-04-1514:01:00
B,1.2,2021-04-1514:02:00
A,1.8,2021-04-1514:03:00 
B,2.5,2021-04-1514:04:00
C,3.8,2021-04-1514:05:00       
C,3.8,2021-04-1514:11:00

最终结果如下：

Flink SQL>SETtable.local-time-zone=UTC; 
Flink SQL>SELECT*FROM MyView4;+-------------------------+-------------------------+-------------------------+------+-----------+|            window_start |              window_end |          window_rowtime | item | max_price |+-------------------------+-------------------------+-------------------------+------+-----------+|2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    A |1.8||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    B |2.5||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    C |3.8|+-------------------------+-------------------------+-------------------------+------+-----------+

Flink SQL>SETtable.local-time-zone=Asia/Shanghai; 
Flink SQL>SELECT*FROM MyView4;+-------------------------+-------------------------+-------------------------+------+-----------+|            window_start |              window_end |          window_rowtime | item | max_price |+-------------------------+-------------------------+-------------------------+------+-----------+|2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    A |1.8||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    B |2.5||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    C |3.8|+-------------------------+-------------------------+-------------------------+------+-----------+

通过上述结果可见，使用 TIMESTAMP（不带时区信息的时间） 进开窗，在 UTC 时区下的计算结果与在 

Asia/Shanghai

时区下计算的窗口开始时间，窗口结束时间和窗口的时间是相同的。

  • TIMESTAMP_LTZ（带时区信息的时间）
Flink SQL>CREATETABLE MyTable3 (
                  item STRING,
                  price DOUBLE,
                  ts BIGINT,-- long 类型的时间戳
                  ts_ltz AS TO_TIMESTAMP_LTZ(ts,3),-- 转为 TIMESTAMP_LTZ 类型的时间戳
                  WATERMARK FOR ts_ltz AS ts_ltz -INTERVAL'10'SECOND)WITH('connector'='socket','hostname'='127.0.0.1','port'='9999','format'='csv');

Flink SQL>CREATEVIEW MyView5 ASSELECT 
                TUMBLE_START(ts_ltz,INTERVAL'10' MINUTES)AS window_start,        
                TUMBLE_END(ts_ltz,INTERVAL'10' MINUTES)AS window_end,
                TUMBLE_ROWTIME(ts_ltz,INTERVAL'10' MINUTES)as window_rowtime,
                item,MAX(price)as max_price
            FROM MyTable3
                GROUPBY TUMBLE(ts_ltz,INTERVAL'10' MINUTES), item;

Flink SQL>DESC MyView5;+----------------+----------------------------+-------+-----+--------+-----------+|           name |type|null|key| extras | watermark |+----------------+----------------------------+-------+-----+--------+-----------+|   window_start |TIMESTAMP(3)|false|||||     window_end |TIMESTAMP(3)|false||||| window_rowtime | TIMESTAMP_LTZ(3)*ROWTIME*|true|||||           item |                     STRING |true|||||      max_price |DOUBLE|true||||+----------------+----------------------------+-------+-----+--------+-----------+

将数据写入 MyTable3：

A,1.1,1618495260000# 对应到 UTC 时区的时间为 2021-04-15 14:01:00
B,1.2,1618495320000# 对应到 UTC 时区的时间为 2021-04-15 14:02:00
A,1.8,1618495380000# 对应到 UTC 时区的时间为 2021-04-15 14:03:00
B,2.5,1618495440000# 对应到 UTC 时区的时间为 2021-04-15 14:04:00
C,3.8,1618495500000# 对应到 UTC 时区的时间为 2021-04-15 14:05:00       
C,3.8,1618495860000# 对应到 UTC 时区的时间为 2021-04-15 14:11:00

最终结果如下：

Flink SQL>SETtable.local-time-zone=UTC; 
Flink SQL>SELECT*FROM MyView5;+-------------------------+-------------------------+-------------------------+------+-----------+|            window_start |              window_end |          window_rowtime | item | max_price |+-------------------------+-------------------------+-------------------------+------+-----------+|2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    A |1.8||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    B |2.5||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:09:59.999|    C |3.8|+-------------------------+-------------------------+-------------------------+------+-----------+

Flink SQL>SETtable.local-time-zone=Asia/Shanghai; 
Flink SQL>SELECT*FROM MyView5;+-------------------------+-------------------------+-------------------------+------+-----------+|            window_start |              window_end |          window_rowtime | item | max_price |+-------------------------+-------------------------+-------------------------+------+-----------+|2021-04-1522:00:00.000|2021-04-1522:10:00.000|2021-04-1522:09:59.999|    A |1.8||2021-04-1522:00:00.000|2021-04-1522:10:00.000|2021-04-1522:09:59.999|    B |2.5||2021-04-1522:00:00.000|2021-04-1522:10:00.000|2021-04-1522:09:59.999|    C |3.8|+-------------------------+-------------------------+-------------------------+------+-----------+

通过上述结果可见，使用 TIMESTAMP_LTZ（带时区信息的时间） 进开窗，在 UTC 时区下的计算结果与在 

Asia/Shanghai

时区下计算的窗口开始时间，窗口结束时间和窗口的时间是不同的，都是按照时区进行格式化的。

5.处理时间和时区应用案例

Flink SQL 定义处理时间属性列是通过 

PROCTIME()

函数来指定的，其返回值类型是 TIMESTAMP_LTZ。

注意：在 Flink 

     1.13 
    
   
  
    1.13 
   
  
1.13 之前，

PROCTIME()

函数返回类型是 TIMESTAMP，返回值是 UTC 时区的时间戳，例如，上海时间显示为 

2021-03-01 12:00:00

时，

PROCTIME()

返回值显示 

2021-03-01 04:00:00

，我们进行使用是错误的。Flink 

     1.13 
    
   
  
    1.13 
   
  
1.13 修复了这个问题，使用 TIMESTAMP_LTZ 作为 

PROCTIME()

的返回类型，这样 Flink 就会自动获取当前时区信息，然后进行处理，不需要用户再进行时区的格式化处理了。

如下案例：

Flink SQL>SETtable.local-time-zone=UTC;
Flink SQL>SELECT PROCTIME();+-------------------------+|              PROCTIME()|+-------------------------+|2021-04-1514:48:31.387|+-------------------------+

Flink SQL>SETtable.local-time-zone=Asia/Shanghai;
Flink SQL>SELECT PROCTIME();+-------------------------+|              PROCTIME()|+-------------------------+|2021-04-1522:48:31.387|+-------------------------+

Flink SQL>CREATETABLE MyTable1 (
                  item STRING,
                  price DOUBLE,
                  proctime as PROCTIME())WITH('connector'='socket','hostname'='127.0.0.1','port'='9999','format'='csv');

Flink SQL>CREATEVIEW MyView3 ASSELECT
                TUMBLE_START(proctime,INTERVAL'10' MINUTES)AS window_start,
                TUMBLE_END(proctime,INTERVAL'10' MINUTES)AS window_end,
                TUMBLE_PROCTIME(proctime,INTERVAL'10' MINUTES)as window_proctime,
                item,MAX(price)as max_price
            FROM MyTable1
                GROUPBY TUMBLE(proctime,INTERVAL'10' MINUTES), item;

Flink SQL>DESC MyView3;+-----------------+-----------------------------+-------+-----+--------+-----------+|           name  |type|null|key| extras | watermark |+-----------------+-----------------------------+-------+-----+--------+-----------+|    window_start |TIMESTAMP(3)|false|||||      window_end |TIMESTAMP(3)|false||||| window_proctime | TIMESTAMP_LTZ(3)*PROCTIME*|false|||||            item |                      STRING |true|||||       max_price |DOUBLE|true||||+-----------------+-----------------------------+-------+-----+--------+-----------+

将数据写入到 MyTable1 中：

> nc -lk 9999
A,1.1
B,1.2
A,1.8
B,2.5
C,3.8

其输出结果如下：

Flink SQL>SETtable.local-time-zone=UTC;
Flink SQL>SELECT*FROM MyView3;+-------------------------+-------------------------+-------------------------+------+-----------+|            window_start |              window_end |          window_procime | item | max_price |+-------------------------+-------------------------+-------------------------+------+-----------+|2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:10:00.005|    A |1.8||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:10:00.007|    B |2.5||2021-04-1514:00:00.000|2021-04-1514:10:00.000|2021-04-1514:10:00.007|    C |3.8|+-------------------------+-------------------------+-------------------------+------+-----------+

Flink SQL>SETtable.local-time-zone=Asia/Shanghai;
Flink SQL>SELECT*FROM MyView3;+-------------------------+-------------------------+-------------------------+------+-----------+|            window_start |              window_end |          window_procime | item | max_price |+-------------------------+-------------------------+-------------------------+------+-----------+|2021-04-1522:00:00.000|2021-04-1522:10:00.000|2021-04-1522:10:00.005|    A |1.8||2021-04-1522:00:00.000|2021-04-1522:10:00.000|2021-04-1522:10:00.007|    B |2.5||2021-04-1522:00:00.000|2021-04-1522:10:00.000|2021-04-1522:10:00.007|    C |3.8|+-------------------------+-------------------------+-------------------------+------+-----------+

通过上述结果可见，使用处理时间进行开窗，在 UTC 时区下的计算结果与在 

Asia/Shanghai

时区下计算的窗口开始时间，窗口结束时间和窗口的时间是不同的，都是按照时区进行格式化的。

6.SQL 时间函数返回在流批任务中的异同

以下函数：

  • LOCALTIME
  • LOCALTIMESTAMP
  • CURRENT_DATE
  • CURRENT_TIME
  • CURRENT_TIMESTAMP
  • NOW()

在 Streaming 模式下这些函数是每条记录都会计算一次，但在 Batch 模式下，只会在 Query 开始时计算一次，所有记录都使用相同的时间结果。

以下时间函数无论是在 Streaming 模式还是 Batch 模式下，都会为每条记录计算一次结果：

  • CURRENT_ROW_TIMESTAMP()
  • PROCTIME()
标签： flink sql 大数据
本文转载自: https://blog.csdn.net/be_racle/article/details/136723893
版权归原作者 G皮T 所有， 如有侵权，请联系我们删除。
登录后发布评论

“【Flink SQL】Flink SQL 基础概念（五）：SQL 时区问题”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

机器人技术在安全领域的重要作用

安全学习记录——网络篇（一）

图腾柱PFC：HP1010为您的电动两轮车之旅提供绿色，高效，安全的动力

单元测试数据库回滚问题

这才是单元测试，之前我们都错了

【C语言基础】：字符函数和字符串函数

ChatGPT-Next-Web SSRF漏洞+XSS漏洞复现(CVE-2023-49785)

文章导航