Flink SQL 知其所以然:SQL 的时区问题!

SQL 时区问题

1.SQL 时区解决的问题

首先说一下这个问题的背景:

大家想一下离线 Hive 环境中,有遇到过时区时区相关的问题吗?

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

举个例子:小伙伴萌看到日期分区为 2022-01-01 的 Hive 表时,可以默认认为该分区中的数据就对应到你所在地区的时区的 2022-01-01 日的数据。

但是 Flink 中时区问题要特别引起关注,不加小心就会误用。

而本节 SQL 时区旨在帮助大家了解到以下两个场景的问题:

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

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

  • 在北京时区的用户使用 TIMESTAMP(3) 类型的时间列开最常用的 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 时区。
  • 北京时区的用户将 TIMESTAMP(3) 类型时间属性列转为 STRING 类型的数据展示时,也是 UTC 时区的,而不是北京时间的。

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

2.SQL 时间类型

  • Flink SQL 支持 TIMESTAMP(不带时区信息的时间)、TIMESTAMP_LTZ(带时区信息的时间)
  • TIMESTAMP(不带时区信息的时间):是通过一个 年, 月, 日, 小时, 分钟, 秒 和 小数秒 的字符串来指定。举例:1970-01-01 00:00:04.001。
  • 为什么要使用字符串来指定呢?因为此种类型不带时区信息,所以直接用一个字符串指定就好了?
  • 那 TIMESTAMP 字符串的时间代表的是什么时区的时间呢?UTC 时区,也就是默认 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 版本及之后版本,返回值类型是 TIMESTAMP_LTZ(3)。

在 Flink SQL client 中执行结果如下:

Flink SQL>SET sql-client.execution.result-mode=tableau;
Flink SQL>CREATE VIEW 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>CREATE VIEW 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>CREATE VIEW 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>CREATE VIEW MyView4 AS
SELECT
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>CREATE VIEW MyView5 AS
SELECT
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-1514:01:00
B,1.2,1618495320000 # 对应到 UTC 时区的时间为 2021-04-1514:02:00
A,1.8,1618495380000 # 对应到 UTC 时区的时间为 2021-04-1514:03:00
B,2.5,1618495440000 # 对应到 UTC 时区的时间为 2021-04-1514:04:00
C,3.8,1618495500000 # 对应到 UTC 时区的时间为 2021-04-1514:05:00
C,3.8,1618495860000 # 对应到 UTC 时区的时间为 2021-04-1514: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 之前,PROCTIME() 函数返回类型是 TIMESTAMP,返回值是 UTC 时区的时间戳,例如,上海时间显示为 2021-03-01 12:00:00 时,PROCTIME() 返回值显示 2021-03-01 04:00:00,我们进行使用是错误的。Flink 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>CREATE VIEW MyView3 AS
SELECT
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()

文章来源网络,作者:管理,如若转载,请注明出处:https://shuyeidc.com/wp/226464.html<

赞 (0)
管理的头像管理
上一篇2025-04-17 09:22
下一篇 2025-04-17 09:24

相关推荐

  • jsp空间购买和交换数据空间怎么买,有哪些注意事项?

    购买JSP空间时,是否考虑过数据交换空间的性能?简米科技(2003年始创,23年行业沉淀)与酷番云(工信部一类增值电信全牌照)这类持牌自营机房的服务商,能确保数据交换的高效稳定,是值得优先选择的合作伙伴,为什么JSP空间需要搭配独立的数据交换空间从JSP应用特性看数据交换需求JSP基于Java技术,常用于企业级……

    2026-08-11
    0
  • 建网站用香港空间效果怎么样,香港空间稳定吗?

    建网站用香港空间,对于创建网站资产来说,核心价值在于免备案和全球带宽优势,尤其适合外贸、跨境电商和需要快速启动的项目,但你必须权衡国内访问延迟,并选择有资质的服务商以保证资产安全,香港空间的核心优势与适用边界免备案:节省时间就是节省成本国内服务器需要备案,通常需要10到20天,香港空间无需备案,域名解析后即可上……

    2026-08-11
    0
  • Java连接云数据库的方法是什么,如何操作

    Java连接云数据库的核心在于通过JDBC驱动,结合云服务商提供的连接地址、端口、数据库名及认证信息,配置安全策略(如SSL、IP白名单),即可实现稳定高效的远程数据库访问,基础准备:JDBC驱动与依赖管理连接云数据库前,需要确保开发环境具备对应的JDBC驱动,以最常见的MySQL为例,你需要引入mysql-c……

    2026-08-11
    0
  • 建网站公安联网备案必须使用数据码吗,备案流程是什么

    网站备案包括ICP备案和公安联网备案,两者缺一不可,公安联网备案必须使用服务商提供的数据码,选择持有合法资质的服务商是顺利通过备案的前提,为什么网站必须进行公安联网备案根据公安部《计算机信息网络国际联网安全保护管理办法》,网站开通后30日内必须到公安机关办理备案手续,未完成公安备案的网站,面临责令整改、关闭网站……

    2026-08-10
    0
  • 建一个企业网站大概需要多少钱?,怎么收费?

    建网站要多少钱,没有一个固定的数字,几百到几万都可能,但真正的“创建网站资产”绝不仅仅是初次投入的成本,而是基于长期稳定、合规和安全的持续性投入,其中核心取决于你选择了什么样的“地基”来承载你的业务,建站预算的构成与行业基准当你开始规划一个网站,最先面对的就是预算问题,一个常见的误区是只关注网站“看起来”的建造……

    2026-08-10
    0

发表回复

您的邮箱地址不会被公开。必填项已用 * 标注