33、Flink 的Table API 和 SQL 中的时区

Flink 系列文章

1、Flink 部署、概念介绍、source、transformation、sink使用示例、四大基石介绍和示例等系列综合文章链接

13、Flink 的table api与sql的基本概念、通用api介绍及入门示例
14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性
15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置（如何处理更新结果）、时态表、流上的join、流上的确定性以及查询配置
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及FileSystem示例（1）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Elasticsearch示例（2）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Kafka示例（3）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及JDBC示例（4）
16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Hive示例（6）
17、Flink 之Table API: Table API 支持的操作（1）
17、Flink 之Table API: Table API 支持的操作（2）
18、Flink的SQL 支持的操作和语法
19、Flink 的Table API 和 SQL 中的内置函数及示例（1）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（2）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（3）
19、Flink 的Table API 和 SQL 中的自定义函数及示例（4）
20、Flink SQL之SQL Client: 不用编写代码就可以尝试 Flink SQL，可以直接提交 SQL 任务到集群上

22、Flink 的table api与sql之创建表的DDL
24、Flink 的table api与sql之Catalogs（介绍、类型、java api和sql实现ddl、java api和sql操作catalog）-1
24、Flink 的table api与sql之Catalogs（java api操作数据库、表）-2
24、Flink 的table api与sql之Catalogs（java api操作视图）-3
24、Flink 的table api与sql之Catalogs（java api操作分区与函数）-4
25、Flink 的table api与sql之函数(自定义函数示例)
26、Flink 的SQL之概览与入门示例
27、Flink 的SQL之SELECT (select、where、distinct、order by、limit、集合操作和去重)介绍及详细示例（1）
27、Flink 的SQL之SELECT (SQL Hints 和 Joins)介绍及详细示例（2）
27、Flink 的SQL之SELECT (窗口函数)介绍及详细示例（3）
27、Flink 的SQL之SELECT (窗口聚合)介绍及详细示例（4）
27、Flink 的SQL之SELECT (Group Aggregation分组聚合、Over Aggregation Over聚合 和 Window Join 窗口关联)介绍及详细示例（5）
27、Flink 的SQL之SELECT (Top-N、Window Top-N 窗口 Top-N 和 Window Deduplication 窗口去重)介绍及详细示例（6）
27、Flink 的SQL之SELECT (Pattern Recognition 模式检测)介绍及详细示例（7）
28、Flink 的SQL之DROP 、ALTER 、INSERT 、ANALYZE 语句
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE（1）
29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE（2）
30、Flink SQL之SQL 客户端（通过kafka和filesystem的例子介绍了配置文件使用-表、视图等）
32、Flink table api和SQL 之用户自定义 Sources & Sinks实现及详细示例
33、Flink 的Table API 和 SQL 中的时区
41、Flink之Hive 方言介绍及详细示例
42、Flink 的table api与sql之Hive Catalog
43、Flink之Hive 读写及详细验证示例
44、Flink之module模块介绍及使用示例和Flink SQL使用hive内置函数及自定义函数详细示例–网上有些说法好像是错误的

文章目录

  • Flink 系列文章
  • 一、时区
    • 1、TIMESTAMP vs TIMESTAMP_LTZ
      • 1）、TIMESTAMP 类型
      • 2）、TIMESTAMP_LTZ 类型
    • 2、时区的作用
      • 1）、确定时间函数的返回值
      • 2）、TIMESTAMP_LTZ 字符串表示
    • 3、时间属性和时区
      • 1）、处理时间和时区
      • 2）、事件时间和时区
        • 1、TIMESTAMP 上的事件时间属性
        • 2、TIMESTAMP_LTZ 上的事件时间属性
    • 4、夏令时支持
    • 5、Batch 模式和 Streaming 模式的区别

本文简单的介绍了Flink 中关于时区的概念，并以具体的示例进行说明。
本文依赖flink、kafka集群能正常使用。
本文分为5个部分，即TIMESTAMP vs TIMESTAMP_LTZ介绍、时区的作用、时区属性与时区、夏令时支持与流批关于时间的处理区别。
本文的示例是在Flink 1.17版本中运行。

一、时区

Flink 为日期和时间提供了丰富的数据类型， 包括 DATE， TIME， TIMESTAMP， TIMESTAMP_LTZ， INTERVAL YEAR TO MONTH， INTERVAL DAY TO SECOND (更多详情请参考 14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性 中的 Date and Time)。 Flink 支持在 session （会话）级别设置时区（更多详情请参考 15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置（如何处理更新结果）、时态表、流上的join、流上的确定性以及查询配置 中的 Planner 配置 table.local-time-zone 部分）。 Flink 对多种时间类型和时区的支持使得跨时区的数据处理变得非常容易。

1、TIMESTAMP vs TIMESTAMP_LTZ

1）、TIMESTAMP 类型

  • TIMESTAMP§ 是 TIMESTAMP§ WITHOUT TIME ZONE 的简写， 精度 p 支持的范围是0-9， 默认是6。
  • TIMESTAMP 用于描述年， 月， 日， 小时， 分钟， 秒 和 小数秒对应的时间戳。
  • TIMESTAMP 可以通过一个字符串来指定，例如：
Flink SQL> SELECT TIMESTAMP '1970-01-01 00:00:04.001';
+-------------------------+
| 1970-01-01 00:00:04.001 |
+-------------------------+

2）、TIMESTAMP_LTZ 类型

  • TIMESTAMP_LTZ（p） 是 TIMESTAMP（p） WITH LOCAL TIME ZONE 的简写， 精度 p 支持的范围是0-9， 默认是6。
  • TIMESTAMP_LTZ 用于描述时间线上的绝对时间点， 使用 long 保存从 epoch 至今的毫秒数， 使用int保存毫秒中的纳秒数。 epoch 时间是从 java 的标准 epoch 时间 1970-01-01T00:00:00Z 开始计算。 在计算和可视化时， 每个 TIMESTAMP_LTZ 类型的数据都是使用的 session （会话）中配置的时区。
  • TIMESTAMP_LTZ 没有字符串表达形式因此无法通过字符串来指定， 可以通过一个 long 类型的 epoch 时间来转化(例如: 通过 Java 来产生一个 long 类型的 epoch 时间 System.currentTimeMillis())
Flink SQL> CREATE VIEW T1 AS SELECT TO_TIMESTAMP_LTZ(4001, 3);
Flink SQL> SET 'table.local-time-zone' = 'UTC';
Flink SQL> SELECT * FROM T1;
+---------------------------+
| TO_TIMESTAMP_LTZ(4001, 3) |
+---------------------------+
|   1970-01-01 00:00:04.001 |
+---------------------------+Flink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai';
Flink SQL> SELECT * FROM T1;
+---------------------------+
| TO_TIMESTAMP_LTZ(4001, 3) |
+---------------------------+
|   1970-01-01 08:00:04.001 |
+---------------------------+
  • TIMESTAMP_LTZ 可以用于跨时区的计算，因为它是一个基于 epoch 的绝对时间点（比如上例中的 4001 毫秒）代表的就是不同时区的同一个绝对时间点。 补充一个背景知识：在同一个时间点， 全世界所有的机器上执行 System.currentTimeMillis() 都会返回同样的值。 (比如上例中的 4001 milliseconds), 这就是绝对时间的定义。

2、时区的作用

本地时区定义了当前 session（会话）所在的时区， 你可以在 Sql client 或者应用程序中配置。

  • java
    代码片段示例
EnvironmentSettings envSetting = EnvironmentSettings.inStreamingMode();TableEnvironment tEnv = TableEnvironment.create(envSetting);// 设置为 UTC 时区tEnv.getConfig().setLocalTimeZone(ZoneId.of("UTC"));// 设置为上海时区tEnv.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));// 设置为 Los_Angeles 时区tEnv.getConfig().setLocalTimeZone(ZoneId.of("America/Los_Angeles"));
  • sql client
-- 设置为 UTC 时区
Flink SQL> SET 'table.local-time-zone' = 'UTC';
[INFO] Execute statement succeed.-- 设置为上海时区
Flink SQL> SET 'table.local-time-zone' = 'America/Los_Angeles';
[INFO] Execute statement succeed.-- 设置为Los_Angeles时区
Flink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai';
[INFO] Execute statement succeed.

session（会话）的时区设置在 Flink SQL 中非常有用， 它的主要用法如下:

1）、确定时间函数的返回值

session （会话）中配置的时区会对以下函数生效。

  • LOCALTIME
  • LOCALTIMESTAMP
  • CURRENT_DATE
  • CURRENT_TIME
  • CURRENT_TIMESTAMP
  • CURRENT_ROW_TIMESTAMP()
  • NOW()
  • PROCTIME()
Flink SQL> SET 'sql-client.execution.result-mode' = 'tableau';
[INFO] Execute statement succeed.Flink SQL> CREATE VIEW MyView1 AS 
> SELECT LOCALTIME, LOCALTIMESTAMP, CURRENT_DATE, CURRENT_TIME, CURRENT_TIMESTAMP, CURRENT_ROW_TIMESTAMP(), NOW(), PROCTIME();
[INFO] Execute statement succeed.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 |     |        |           |
|            EXPR$5 |            TIMESTAMP_LTZ(3) | FALSE |     |        |           |
|            EXPR$6 |            TIMESTAMP_LTZ(3) | FALSE |     |        |           |
|            EXPR$7 | TIMESTAMP_LTZ(3) *PROCTIME* | FALSE |     |        |           |
+-------------------+-----------------------------+-------+-----+--------+-----------+
8 rows in setFlink SQL> SET 'table.local-time-zone' = 'UTC';
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView1;
+----+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| op | LOCALTIME |          LOCALTIMESTAMP | CURRENT_DATE | CURRENT_TIME |       CURRENT_TIMESTAMP |                  EXPR$5 |                  EXPR$6 |                  EXPR$7 |
+----+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| +I |  06:52:14 | 2023-11-10 06:52:14.144 |   2023-11-10 |     06:52:14 | 2023-11-10 06:52:14.144 | 2023-11-10 06:52:14.144 | 2023-11-10 06:52:14.144 | 2023-11-10 06:52:14.145 |
+----+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
Received a total of 1 rowFlink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai';
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView1;
+----+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| op | LOCALTIME |          LOCALTIMESTAMP | CURRENT_DATE | CURRENT_TIME |       CURRENT_TIMESTAMP |                  EXPR$5 |                  EXPR$6 |                  EXPR$7 |
+----+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
| +I |  14:52:52 | 2023-11-10 14:52:52.305 |   2023-11-10 |     14:52:52 | 2023-11-10 14:52:52.305 | 2023-11-10 14:52:52.305 | 2023-11-10 14:52:52.305 | 2023-11-10 14:52:52.305 |
+----+-----------+-------------------------+--------------+--------------+-------------------------+-------------------------+-------------------------+-------------------------+
Received a total of 1 row

2）、TIMESTAMP_LTZ 字符串表示

当一个 TIMESTAMP_LTZ 值转为 string 格式时， session 中配置的时区会生效。 例如打印这个值，将类型强制转化为 STRING 类型， 将类型强制转换为 TIMESTAMP ，将 TIMESTAMP 的值转化为 TIMESTAMP_LTZ 类型：

Flink SQL> CREATE VIEW MyView2 AS 
> SELECT TO_TIMESTAMP_LTZ(4001, 3) AS ltz, TIMESTAMP '1970-01-01 00:00:01.001'  AS ntz;
[INFO] Execute statement succeed.Flink SQL> DESC MyView2;
+------+------------------+-------+-----+--------+-----------+
| name |             type |  null | key | extras | watermark |
+------+------------------+-------+-----+--------+-----------+
|  ltz | TIMESTAMP_LTZ(3) |  TRUE |     |        |           |
|  ntz |     TIMESTAMP(3) | FALSE |     |        |           |
+------+------------------+-------+-----+--------+-----------+
2 rows in setFlink SQL> SET 'table.local-time-zone' = 'UTC';
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView2;
+----+-------------------------+-------------------------+
| op |                     ltz |                     ntz |
+----+-------------------------+-------------------------+
| +I | 1970-01-01 00:00:04.001 | 1970-01-01 00:00:01.001 |
+----+-------------------------+-------------------------+
Received a total of 1 rowFlink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai';
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView2;
+----+-------------------------+-------------------------+
| op |                     ltz |                     ntz |
+----+-------------------------+-------------------------+
| +I | 1970-01-01 08:00:04.001 | 1970-01-01 00:00:01.001 |
+----+-------------------------+-------------------------+
Received a total of 1 rowFlink SQL> CREATE VIEW MyView3 AS 
> SELECT ltz, CAST(ltz AS TIMESTAMP(3)), CAST(ltz AS STRING), ntz, CAST(ntz AS TIMESTAMP_LTZ(3)) FROM MyView2;
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView3;
+----+-------------------------+-------------------------+--------------------------------+-------------------------+-------------------------+
| op |                     ltz |                  EXPR$1 |                         EXPR$2 |                     ntz |                  EXPR$4 |
+----+-------------------------+-------------------------+--------------------------------+-------------------------+-------------------------+
| +I | 1970-01-01 08:00:04.001 | 1970-01-01 08:00:04.001 |        1970-01-01 08:00:04.001 | 1970-01-01 00:00:01.001 | 1970-01-01 00:00:01.001 |
+----+-------------------------+-------------------------+--------------------------------+-------------------------+-------------------------+
Received a total of 1 row

3、时间属性和时区

更多时间属性相关的详细介绍， 请参考：15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置（如何处理更新结果）、时态表、流上的join、流上的确定性以及查询配置 中的时间属性配置部分。

1）、处理时间和时区

Flink SQL 使用函数 PROCTIME() 来定义处理时间属性， 该函数返回的类型是 TIMESTAMP_LTZ 。

在 Flink1.13 之前， PROCTIME() 函数返回的类型是 TIMESTAMP ， 返回值是UTC时区下的 TIMESTAMP 。
例如： 当上海的时间为 2021-11-11 12:00:00 时， PROCTIME() 显示的时间却是错误的 2021-11-11 04:00:00 。 这个问题在 Flink 1.13 中修复了， 因此用户不用再去处理时区的问题了。

PROCTIME() 返回的是本地时区的时间， 使用 TIMESTAMP_LTZ 类型也可以支持夏令时时间。

Flink SQL> SET 'table.local-time-zone' = 'UTC';
[INFO] Execute statement succeed.Flink SQL> SELECT PROCTIME();
+----+-------------------------+
| op |                  EXPR$0 |
+----+-------------------------+
| +I | 2023-11-10 06:59:30.998 |
+----+-------------------------+
Received a total of 1 rowFlink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai';
[INFO] Execute statement succeed.Flink SQL> SELECT PROCTIME();
+----+-------------------------+
| op |                  EXPR$0 |
+----+-------------------------+
| +I | 2023-11-10 14:59:54.031 |
+----+-------------------------+
Received a total of 1 rowFlink SQL> CREATE TABLE MyTable1 (
>                   item STRING,
>                   price DOUBLE,
>                   proctime as PROCTIME()
> ) WITH (
>   'connector' = 'kafka',
>   'topic' = 'MyTable1',
>   'properties.bootstrap.servers' = '192.168.10.41:9092,192.168.10.42:9092,192.168.10.43:9092',
>   'properties.group.id' = 'testGroup',
>   'scan.startup.mode' = 'earliest-offset',
>   'format' = 'csv'
> );
[INFO] Execute statement succeed.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
>                 GROUP BY TUMBLE(proctime, INTERVAL '10' MINUTES), item;
[INFO] Execute statement succeed.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 |     |        |           |
+-----------------+-----------------------------+-------+-----+--------+-----------+
5 rows in set

在终端执行以下命令写入数据到 MyTable1 ：

[alanchan@server1 bin]$ kafka-console-producer.sh --broker-list server1:9092 --topic MyTable1
>A,1.1
>B,1.2
>A,1.8
>B,2.5
>C,3.8
>
Flink SQL> SET 'table.local-time-zone' = 'UTC';
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView3;
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| op |            window_start |              window_end |         window_proctime |                           item |                      max_price |
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| +I | 2023-11-10 07:10:00.000 | 2023-11-10 07:20:00.000 | 2023-11-10 07:20:00.000 |                              A |                            1.8 |
| +I | 2023-11-10 07:10:00.000 | 2023-11-10 07:20:00.000 | 2023-11-10 07:20:00.000 |                              C |                            3.8 |
| +I | 2023-11-10 07:10:00.000 | 2023-11-10 07:20:00.000 | 2023-11-10 07:20:00.001 |                              B |                            2.5 |
received a total of 3 rows

Flink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai';
[INFO] Execute statement succeed.

相比在 UTC 时区下的计算结果， 在 Asia/Shanghai 时区下计算的窗口开始时间， 窗口结束时间和窗口处理时间是不同的。

Flink SQL> SELECT * FROM MyView3;
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| op |            window_start |              window_end |         window_proctime |                           item |                      max_price |
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| +I | 2023-11-10 15:10:00.000 | 2023-11-10 15:20:00.000 | 2023-11-10 15:20:00.000 |                              A |                            1.8 |
| +I | 2023-11-10 15:10:00.000 | 2023-11-10 15:20:00.000 | 2023-11-10 15:20:00.000 |                              C |                            3.8 |
| +I | 2023-11-10 15:10:00.000 | 2023-11-10 15:20:00.000 | 2023-11-10 15:20:00.001 |                              B |                            2.5 |
received a total of 3 rows

处理时间窗口是不确定的， 每次运行都会返回不同的窗口和聚合结果。 以上的示例只用于说明时区如何影响处理时间窗口。

2）、事件时间和时区

Flink 支持在 TIMESTAMP 列和 TIMESTAMP_LTZ 列上定义时间属性。

1、TIMESTAMP 上的事件时间属性

如果 source 中的时间用于表示年-月-日-小时-分钟-秒， 通常是一个不带时区的字符串，
例如: 2023-11-13 08:13:40.564。 推荐在 TIMESTAMP 列上定义事件时间属性。

  • 准备测试环境，即表、视图和数据
Flink SQL> CREATE TABLE MyTable2 (
>   item STRING,
>   price DOUBLE,
>   ts TIMESTAMP(3), -- TIMESTAMP data type
>   WATERMARK FOR ts AS ts - INTERVAL '10' SECOND
>    ) WITH (
>   'connector' = 'kafka',
>   'topic' = 'MyTable2',
>   'properties.bootstrap.servers' = '192.168.10.41:9092,192.168.10.42:9092,192.168.10.43:9092',
>   'properties.group.id' = 'testGroup',
>   'scan.startup.mode' = 'earliest-offset',
>   'format' = 'csv'
> );
[INFO] Execute statement succeed.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
> GROUP BY TUMBLE(ts, INTERVAL '10' MINUTES), item;
[INFO] Execute statement succeed.Flink SQL> DESC MyView4;
+----------------+------------------------+-------+-----+--------+-----------+
|           name |                   type |  null | key | extras | watermark |
+----------------+------------------------+-------+-----+--------+-----------+
|   window_start |           TIMESTAMP(3) | FALSE |     |        |           |
|     window_end |           TIMESTAMP(3) | FALSE |     |        |           |
| window_rowtime | TIMESTAMP(3) *ROWTIME* |  TRUE |     |        |           |
|           item |                 STRING |  TRUE |     |        |           |
|      max_price |                 DOUBLE |  TRUE |     |        |           |
+----------------+------------------------+-------+-----+--------+-----------+
5 rows in set

在终端执行以下命令用于写入数据到 MyTable2 ：

[alanchan@server1 bin]$ kafka-console-producer.sh --broker-list server1:9092 --topic alan_MyTable2
>A,1.1,2023-11-13 08:21:00
>B,1.2,2023-11-13 08:22:00
>A,1.8,2023-11-13 08:23:00
>B,2.5,2023-11-13 08:24:00
>C,3.8,2023-11-13 08:25:00
>C,3.8,2023-11-13 08:41:00
  • 查看UTC与Asia/Shanghai的查询结果
Flink SQL> SET 'table.local-time-zone' = 'UTC'; 
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView4;
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| op |            window_start |              window_end |          window_rowtime |                           item |                      max_price |
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| +I | 2023-11-13 08:20:00.000 | 2023-11-13 08:30:00.000 | 2023-11-13 08:29:59.999 |                              A |                            1.8 |
| +I | 2023-11-13 08:20:00.000 | 2023-11-13 08:30:00.000 | 2023-11-13 08:29:59.999 |                              B |                            2.5 |
| +I | 2023-11-13 08:20:00.000 | 2023-11-13 08:30:00.000 | 2023-11-13 08:29:59.999 |                              C |                            3.8 |

相比在 UTC 时区下的计算结果， 在 Asia/Shanghai 时区下计算的窗口开始时间， 窗口结束时间和窗口的 rowtime 是相同的。

Flink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai'; 
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView4;
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| op |            window_start |              window_end |          window_rowtime |                           item |                      max_price |
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| +I | 2023-11-13 08:20:00.000 | 2023-11-13 08:30:00.000 | 2023-11-13 08:29:59.999 |                              A |                            1.8 |
| +I | 2023-11-13 08:20:00.000 | 2023-11-13 08:30:00.000 | 2023-11-13 08:29:59.999 |                              B |                            2.5 |
| +I | 2023-11-13 08:20:00.000 | 2023-11-13 08:30:00.000 | 2023-11-13 08:29:59.999 |                              C |                            3.8 |
2、TIMESTAMP_LTZ 上的事件时间属性

如果源数据中的时间为一个 epoch 时间， 通常是一个 long 值， 例如: 1618989564564 ，推荐将事件时间属性定义在 TIMESTAMP_LTZ 列上。

  • 准备测试环境，即准备表、视图和数据
Flink SQL> CREATE TABLE MyTable3 (
>  item STRING,
>  price DOUBLE,
>  ts BIGINT, -- long time value in epoch milliseconds
>  ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
>  WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '10' SECOND
> ) WITH (
>   'connector' = 'kafka',
>   'topic' = 'alan_MyTable3',
>   'properties.bootstrap.servers' = '192.168.10.41:9092,192.168.10.42:9092,192.168.10.43:9092',
>   'properties.group.id' = 'testGroup',
>   'scan.startup.mode' = 'earliest-offset',
>   'format' = 'csv'
> );
[INFO] Execute statement succeed.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
> GROUP BY TUMBLE(ts_ltz, INTERVAL '10' MINUTES), item;
[INFO] Execute statement succeed.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 |     |        |           |
+----------------+----------------------------+-------+-----+--------+-----------+
5 rows in set

在终端执行以下命令用于写入数据到 MyTable3 ：

[alanchan@server1 bin]$ kafka-console-producer.sh --broker-list server1:9092 --topic alan_MyTable3
>A,1.1,1699836971034 # The corresponding utc timestamp is 2023-11-13 08:56:xx
>B,1.2,1699837031044 # The corresponding utc timestamp is 2023-11-13 08:57:xx
>A,1.8,1699837091052 # The corresponding utc timestamp is 2023-11-13 08:58:xx
>B,2.5,1699837091052 # The corresponding utc timestamp is 2023-11-13 08:59:xx
>C,3.8,1699837211069 # The corresponding utc timestamp is 2023-11-13 09:00:xx
>C,3.8,1699837271070 # The corresponding utc timestamp is 2023-11-13 09:01:xx
  • 查看UTC与Asia/Shanghai的查询结果
Flink SQL> SET 'table.local-time-zone' = 'UTC'; 
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView5;
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| op |            window_start |              window_end |          window_rowtime |                           item |                      max_price |
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| +I | 2023-11-13 00:50:00.000 | 2023-11-13 01:00:00.000 | 2023-11-13 00:59:59.999 |                              A |                            1.8 |
| +I | 2023-11-13 00:50:00.000 | 2023-11-13 01:00:00.000 | 2023-11-13 00:59:59.999 |                              B |                            2.5 |

相比在 UTC 时区下的计算结果， 在 Asia/Shanghai 时区下计算的窗口开始时间， 窗口结束时间和窗口的 rowtime 是不同的。

Flink SQL> SET 'table.local-time-zone' = 'Asia/Shanghai'; 
[INFO] Execute statement succeed.Flink SQL> SELECT * FROM MyView5;
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| op |            window_start |              window_end |          window_rowtime |                           item |                      max_price |
+----+-------------------------+-------------------------+-------------------------+--------------------------------+--------------------------------+
| +I | 2023-11-13 08:50:00.000 | 2023-11-13 09:00:00.000 | 2023-11-13 08:59:59.999 |                              A |                            1.8 |
| +I | 2023-11-13 08:50:00.000 | 2023-11-13 09:00:00.000 | 2023-11-13 08:59:59.999 |                              B |                            2.5 |

4、夏令时支持

Flink SQL支持在 TIMESTAMP_LTZ列上定义时间属性， 基于这一特征，Flink SQL 在窗口中使用 TIMESTAMP 和 TIMESTAMP_LTZ 类型优雅地支持了夏令时。

Flink 使用时间戳的字符格式来分割窗口并通过每条记录对应的 epoch 时间来分配窗口。 这意味着 Flink 窗口开始时间和窗口结束时间使用的是 TIMESTAMP 类型（例如: TUMBLE_START 和 TUMBLE_END）， 窗口的时间属性使用的是 TIMESTAMP_LTZ 类型（例如: TUMBLE_PROCTIME， TUMBLE_ROWTIME）。

给定一个 tumble window示例， 在 Los_Angeles 时区下夏令时从 2021-03-14 02:00:00 开始：

long epoch1 = 1615708800000L; // 2021-03-14 00:00:00
long epoch2 = 1615712400000L; // 2021-03-14 01:00:00
long epoch3 = 1615716000000L; // 2021-03-14 03:00:00, 手表往前拨一小时，跳过 (2021-03-14 02:00:00)
long epoch4 = 1615719600000L; // 2021-03-14 04:00:00

在 Los_angele 时区下， tumble window [2021-03-14 00:00:00, 2021-03-14 00:04:00] 将会收集3个小时的数据， 在其他非夏令时的时区下将会收集4个小时的数据，用户只需要在 TIMESTAMP_LTZ 列上声明时间属性即可。

Flink 的所有窗口（如 Hop window， Session window， Cumulative window）都会遵循这种方式， Flink SQL 中的所有操作都很好地支持了 TIMESTAMP_LTZ 类型，因此Flink可以非常优雅的支持夏令时。

5、Batch 模式和 Streaming 模式的区别

以下函数：

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

Flink 会根据执行模式来进行不同计算，在 Streaming 模式下这些函数是每条记录都会计算一次，但在 Batch 模式下，只会在 query 开始时计算一次，所有记录都使用相同的结果。

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

  • CURRENT_ROW_TIMESTAMP()
  • PROCTIME()

以上，简单的介绍了Flink 中关于时区的概念，并以具体的示例进行说明。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/141707.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

MATLAB Simulink和SMART PLC水箱液位高度PID控制(联合仿真)

MATLAB Simulink和SMART PLC水箱液位高度PID控制(联合仿真)

SMART PLC 向导PID的详细介绍请查看下面文章链接： S7-200 SMART PLC PID向导详细介绍(如何实现P、PD、PID控制器)-CSDN博客文章浏览阅读1k次。这篇博客主要介绍SMART PLC PID向导的使用，PID控制相关的其它内容请查看专栏系列文章，常用链接如下：SMART PLC PID负压控制（过程…
阅读更多...
海外ASO优化之谷歌商店的评论优化

海外ASO优化之谷歌商店的评论优化

应用商店中的评分和评论，显示我们的应用程序的受欢迎程度以及用户对该应用程序的看法。评分和评论是以前或者是现在的用户分享的经验和公开的反馈。 1、提高应用评分评论。 高评分的应用可以从应用商店内的搜索流量中获得更多的点击量，通过推荐和推荐获…
阅读更多...
SMART PLC 和S7-1200PLC MODBUSTCP通信速度测试

SMART PLC 和S7-1200PLC MODBUSTCP通信速度测试

SMART PLC MODBUSTCP通信详细介绍请参看下面文章链接： S7-200SMART PLC ModbusTCP通信（多服务器多从站轮询）_matlab sumilink 多个modbustcp读写_RXXW_Dor的博客-CSDN博客文章浏览阅读6.4k次，点赞5次，收藏10次。MBUS_CLIENT作为MODBUS TCP客户端通过S7-200 SMART CPU上的…
阅读更多...
Python开源项目VQFR——人脸重建（Face Restoration），模糊清晰、划痕修复及黑白上色的实践

Python开源项目VQFR——人脸重建（Face Restoration），模糊清晰、划痕修复及黑白上色的实践

Python Anaconda 的安装等请参阅： Python开源项目CodeFormer——人脸重建（Face Restoration），模糊清晰、划痕修复及黑白上色的实践https://blog.csdn.net/beijinghorn/article/details/134334021 VQFR也是 腾讯 LAB 的作品&…
阅读更多...
【uniapp】确认弹出框，选择确定和取消

【uniapp】确认弹出框，选择确定和取消

代码如下&#xff1a; <view style"display: flex; justify-content: space-around;"><button class"button" click"submit">t提交</button> </view>submit(){let thatthisuni.showModal({title: 提示&#xff1a;,con…
阅读更多...
NSS [HUBUCTF 2022 新生赛]checkin

NSS [HUBUCTF 2022 新生赛]checkin

NSS [HUBUCTF 2022 新生赛]checkin 判断条件是if ($data_unserialize[username]$username&&$data_unserialize[password]$password)&#xff0c;满足则给我们flag。正常思路来说&#xff0c;我们要使序列化传入的username和password等于代码中的两个同名变量&#xff0…
阅读更多...
【工程部署】在RK3588上部署OCR(文字检测识别)（DBNet+CRNN）

【工程部署】在RK3588上部署OCR(文字检测识别)（DBNet+CRNN）

硬件平台&#xff1a; 1、firefly安装Ubuntu系统的RK3588&#xff1b; 2、安装Windows系统的电脑一台&#xff0c;其上安装Ubuntu18.04系统虚拟机。 参考手册&#xff1a;《00-Rockchip_RKNPU_User_Guide_RKNN_API_V1.3.0_CN》 《RKNN Toolkit Lite2 用户使用指南》 1、文…
阅读更多...
mapreduce-maven--30.串联所有单词的字串

mapreduce-maven--30.串联所有单词的字串

项目对象模型&#xff08;Project Object Model&#xff0c;POM&#xff09;&#xff1a;Maven使用POM文件来描述项目的结构、依赖和构建设置。POM是一个XML文件&#xff0c;位于项目根目录下&#xff0c;并包含项目的基本信息、构建设置、依赖管理等。 依赖管理&#xff1a;M…
阅读更多...
Unity DOTS系列之System中如何使用SystemAPI.Query迭代数据

Unity DOTS系列之System中如何使用SystemAPI.Query迭代数据

最近DOTS发布了正式的版本, 我们来分享一下System中如何基于SystemAPI.Query来迭代World中的数据&#xff0c;方便大家上手学习掌握Unity DOTS开发。 SystemAPI.Query的使用 System有两种&#xff0c;一种是Unmanaged 的ISystem,一种是managed 的SystemBase,这两种System都可…
阅读更多...
LeetCode106. Construct Binary Tree from Inorder and Postorder Traversal

LeetCode106. Construct Binary Tree from Inorder and Postorder Traversal

文章目录 一、题目二、题解 一、题目 Given two integer arrays inorder and postorder where inorder is the inorder traversal of a binary tree and postorder is the postorder traversal of the same tree, construct and return the binary tree. Example 1: Input: …
阅读更多...
段的概念_重定位的引入

段的概念_重定位的引入

段的概念 代码段、只读数据段、可读可写数据段、BSS段。 char g_Char A; //可读可写&#xff0c;不能放在ROM上&#xff0c;应该放在RAM里 const char g_Char2 B; //只读变量&#xff0c;可以放在ROM上 int g_A 0; //初始值为0&#xff0c;没有必要浪费空间 int g_B; //没…
阅读更多...
力扣-66. 加一

力扣-66. 加一

文章目录 解题思路代码 解题思路 根据题目可以得出此题一共有3种需要特殊判断的地方。1.数组元素的最后一个元素是否为9&#xff0c;如果是那就将此处元素置0&#xff0c;如果不是那就此处元素加1。按照这个逻辑将digits数组重新赋值。2.判断重新赋值的digits数组的首元素是否…
阅读更多...
Python爬虫从基础到入门：找数据接口

Python爬虫从基础到入门：找数据接口

Python爬虫从基础到入门：找数据接口 1. 怎样判断抓取的数据是动态生成的2. 用requests模块访问，然后用解析模块解析数据3. 总结1. 怎样判断抓取的数据是动态生成的 请参考文章：Python爬虫从基础到入门：认识爬虫 第3点所讲。 这里用我的CSDN个人主页举例。 可以说这部分下的…
阅读更多...
EukCC2评估真核生物MGAs质量

EukCC2评估真核生物MGAs质量

文章目录 简介和原理Install配置数据库使用单个bin包含bins的目录Bin merging示例数据集自测数据更多参数 参考 简介和原理 EukCC2是一个基于python编写的用于评估真核生物MAGs完整度和污染度的软件。可以评估binning后的单个bin或者bins目录。 其原理是基于动态变化的单拷贝标…
阅读更多...
第4关：非递归实现二叉树左右子树交换

第4关：非递归实现二叉树左右子树交换

任务描述相关知识 栈的基本操作二叉树后序遍历编程要求测试说明 任务描述 本关任务&#xff1a;给定一棵二叉树&#xff0c;使用非递归的方式实现二叉树左右子树交换&#xff0c;并输出后序遍历结果。 相关知识 为了完成本关任务&#xff0c;你需要掌握&#xff1a;1.栈的基…
阅读更多...
bug：Junit5报错，@SpringBootTest没有运行

bug：Junit5报错，@SpringBootTest没有运行

1、首先解决Junit5报错 java.lang.NoClassDefFoundError: org/junit/platform/launcher/core/LauncherFactory 添加依赖 implementation org.junit.platform:junit-platform-launcher:1.8.2java.lang.IllegalArgumentException: Error: test loader org.eclipse.jdt.internal.…
阅读更多...
springboot集成kafka详解

springboot集成kafka详解

文章目录 1、kafka部署&#xff1a;&#xff08;1&#xff09;先创建一个网络&#xff1a;&#xff08;2&#xff09;安装zookeeper&#xff0c;kafka依赖zookeeper所以需要先安装zookeeper&#xff1a;&#xff08;3&#xff09;安装Kafka&#xff1a;参数解释&#xff1a; &a…
阅读更多...
c++ jthread 使用详解

c++ jthread 使用详解

c jthread 使用详解 std::jthread c20 头文件 #include <thread>。对 std::thread 的改进&#xff0c;旨在提供更好的线程管理和异常处理&#xff0c;除了拥有和 std::thread 完全相同的功能以及 API 之外&#xff0c;还增加了以下功能&#xff1a; 可停止&#xff1a;…
阅读更多...
无人地磅称重系统|自助过磅 料仓联动 自助卸料

无人地磅称重系统|自助过磅 料仓联动 自助卸料

上海思伟无人地磅系统 自助过磅、 自助卸料 、料仓联动 智能、省人、安全 无人监管过磅 对地磅及其相关的所有硬件进行配置和管理&#xff1b; 支持红外、道闸、车牌识别、AI分析、拍照存档、LED语音播报一体机等设备&#xff1b; 实现稳定可靠的无人监管称重功能&#xf…
阅读更多...
云服务器哪家强？阿里云双十一2核2G配置3M带宽仅99元/年！

云服务器哪家强？阿里云双十一2核2G配置3M带宽仅99元/年！

阿里云作为国内知名的云计算服务提供商&#xff0c;每年的双11都会推出各种优惠活动和促销策略。在今年的双11期间&#xff0c;阿里云推出了多种选择的云服务器&#xff0c;其中两款备受用户关注&#xff1a;轻量服务器2核2G3M带宽优惠价87元一年和经济型e实例2核2G配置3M带宽9…
阅读更多...
最新文章

Copyright @ 2022~2023