踩坑记 | Flink 天级别窗口中存在的时区问题
本系列每篇文章都是从一些实际的 case 出发，分析一些生产环境中经常会遇到的问题，抛砖引玉，以帮助小伙伴们解决一些实际问题。本文介绍 Flink 时间以及时区问题，分析了在天级别的窗口时会遇到的时区问题，如果对小伙伴有帮助的话，欢迎点赞 + 再看~
本文主要分为两部分：
第一部分（第 1 - 3 节）的分析主要针对 flink，分析了 flink 天级别窗口的中存在的时区问题以及解决方案。
第二部分（第 4 节）的分析可以作为所有时区问题的分析思路，主要以解决方案中的时区偏移量为什么是加 8 小时为案例做了通用的深度解析。
为了让读者能对本文探讨的问题有一个大致了解，本文先给出问题 sql，以及解决方案。后文给出详细的分析~
1.问题以及解决方案
问题 sql
sql 很简单，用来统计当天累计 uv。
--------------- 伪代码 ---------------
INSERT INTO
kafka_sink_table
SELECT
-- 窗口开始时间
CAST(
TUMBLE_START(proctime, INTERVAL '1' DAY) AS bigint
) AS window_start,
-- 当前记录处理的时间
cast(max(proctime) AS BIGINT) AS current_ts,
-- 每个桶内的 uv
count(DISTINCT id) AS part_daily_full_uv
FROM
kafka_source_table
GROUP BY
mod(id, bucket_number),
-- bucket_number 为常数，根据具体场景指定具体数值
TUMBLE(proctime, INTERVAL '1' DAY)
--------------- 伪代码 ---------------
你是否能一眼看出这个 sql 所存在的问题？（PS：数据源以及数据汇时区都为东八区）
没错，天级别窗口所存在的时区问题，即这段代码统计的不是楼主所在东八区一整天数据的 uv，这段代码统计的一整天的范围在东八区是第一天早 8 点至第二天早 8 点。
解决方案
楼主目前所处时区为东八区，解决方案如下：
--------------- 伪代码 ---------------
CREATE VIEW view_table AS
SELECT
id,
-- 通过注入时间解决
-- 加上东八区的时间偏移量，设置注入时间为时间戳列
CAST(CURRENT_TIMESTAMP AS BIGINT) * 1000 + 8 * 60 * 60 * 1000 as ingest_time
FROM
source_table;
INSERT INTO
target_table
SELECT
CAST(
TUMBLE_START(ingest_time, INTERVAL '1' DAY) AS bigint
) AS window_start,
cast(max(ingest_time) AS BIGINT) - 8 * 3600 * 1000 AS current_ts,
count(DISTINCT id) AS part_daily_full_uv
FROM
view_table
GROUP BY
mod(id, 1024),
-- 根据注入时间划分天级别窗口
TUMBLE(ingest_time, INTERVAL '1' DAY)
--------------- 伪代码 ---------------
通过上述方案，就可以将统计的数据时间范围调整为东八区的今日 0 点至明日 0 点。下文详细说明整个需求场景以及解决方案的实现和分析过程。
2.需求场景以及实现方案
需求场景
coming，需求场景比较简单，就是消费上游的一个埋点日志数据源，根据埋点中的 id 统计当天 0 点至当前时刻的累计 uv，按照分钟级别产出到下游 OLAP 引擎中进行简单的聚合，最后在 BI 看板进行展示，没有任何维度字段（感动到哭😭）。
数据链路以及组件选型
客户端用户行为埋点日志 -> logServer -> kafka -> flink（sql） -> kafka -> druid -> BI 看板。
实现方案以及具体的实现方式很多，这次使用的是 sql API。
flink sql schema
source 和 sink 表 schema 如下（只保留关键字段）：
--------------- 伪代码 ---------------
CREATE TABLE kafka_sink_table (
-- 天级别窗口开始时间
window_start BIGINT,
-- 当前记录处理的时间
current_ts BIGINT,
-- 每个桶内的 uv（处理过程对 id 进行了分桶）
part_daily_full_uv BIGINT
) WITH (
-- ...
);
CREATE TABLE kafka_source_table (
-- ...
-- 需要进行 uv 计算的 id
id BIGINT,
-- 处理时间
proctime AS PROCTIME()
) WITH (
-- ...
);
--------------- 伪代码 ---------------
flink sql transform
--------------- 伪代码 ---------------
INSERT INTO
kafka_sink_table
SELECT
-- 窗口开始时间
CAST(
TUMBLE_START(proctime, INTERVAL '1' DAY) AS bigint
) AS window_start,
-- 当前记录处理的时间
cast(max(proctime) AS BIGINT) AS current_ts,
-- 每个桶内的 uv
count(DISTINCT id) AS part_daily_full_uv
FROM
kafka_source_table
GROUP BY
mod(id, bucket_number),
-- bucket_number 为常数，根据具体场景指定具体数值
TUMBLE(proctime, INTERVAL '1' DAY)
--------------- 伪代码 ---------------
使用 early-fire 机制（同 DataStream API 中的 ContinuousProcessingTimeTrigger），并设定触发间隔为 60 s。
在上述实现 sql 中，我们对 id 进行了分桶，那么每分钟输出的数据条数即为 bucket_number 条，最终在 druid 中按照分钟粒度将所有桶的数据进行 sum 聚合，即可得到从当天 0 点累计到当前分钟的全量 uv。
时区问题
激情场景还原：
头文字 ∩ 技术小哥哥：使用 sql，easy game，闲坐摸鱼...
头文字 ∩ 技术小哥哥：等到 00:00 时，发现指标还在不停地往上涨，难道是 sql 逻辑错了，不应该啊，试过分钟，小时级别窗口都木有这个问题
头文字 ∩ 技术小哥哥：抠头ing，算了，稍后再分析这个问题吧，现在还有正事要干😏
头文字 ∩ 技术小哥哥：到了早上，瞅了一眼配置的时间序列报表，发现在 08:00 点的时候指标归零，重新开始累计。想法一闪而过，东八区？（当时为啥没 format 下 sink 数据中的 window_start...）
3.问题定位
问题说明
flink 在使用时间的这个概念的时候是基于 java 时间纪元（即格林威治 1970/01/01 00:00:00，也即 Unix 时间戳为 0）概念的，窗口对齐以及触发也是基于 java 时间纪元。
问题场景复现
可以通过直接查看 sink 数据的 window_start 得出上述结论。
但为了还原整个过程，我们按照如下 source 和 sink 数据进行整个问题的复现：
source 数据如下：
| id | proctime | proctime UTC + 0（格林威治） 格式化时间 | proctime UTC + 8（北京） 格式化时间 |
|---|---|---|---|
| 1 | 1599091140000 | 2020/09/02 23:59:00 | 2020/09/03 07:59:00 |
| 2 | 1599091140000 | 2020/09/02 23:59:00 | 2020/09/03 07:59:00 |
| 3 | 1599091140000 | 2020/09/02 23:59:00 | 2020/09/03 07:59:00 |
| 1 | 1599091200000 | 2020/09/03 00:00:00 | 2020/09/03 08:00:00 |
| 2 | 1599091200000 | 2020/09/03 00:00:00 | 2020/09/03 08:00:00 |
| 3 | 1599091260000 | 2020/09/03 00:01:00 | 2020/09/03 08:01:00 |
sink 数据（为了方便理解，直接按照 druid 聚合之后的数据展示）：
| window_start | current_ts | part_daily_full_uv | window_start UTC + 8（北京） 格式化时间 | current_ts UTC + 8（北京） 格式化时间 |
|---|---|---|---|---|
| 1599004800000 | 1599091140000 | 3 | 2020/09/02 08:00:00 | 2020/09/03 07:59:00 |
| 1599091200000 | 1599091200000 | 2 | 2020/09/03 08:00:00 | 2020/09/03 08:00:00 |
| 1599091200000 | 1599091260000 | 3 | 2020/09/03 08:00:00 | 2020/09/03 08:01:00 |
从上述数据可以发现，天级别窗口开始时间在 UTC + 8（北京）的时区是每天早上 8 点，即 UTC + 0（格林威治）的凌晨 0 点。
下文先给出解决方案，然后详细解析各个时间以及时区概念~
解决方案
- 框架层面解决：Blink Planner 支持时区设置
- sql层面解决：从 sql 实现层面给出解决方案
sql 层面解决方案
--------------- 伪代码 ---------------
CREATE VIEW view_table AS
SELECT
id,
-- 通过注入时间解决
-- 加上东八区的时间偏移量，设置注入时间为时间戳列
CAST(CURRENT_TIMESTAMP AS BIGINT) * 1000 + 8 * 60 * 60 * 1000 as ingest_time
FROM
source_table;
INSERT INTO
target_table
SELECT
CAST(
TUMBLE_START(ingest_time, INTERVAL '1' DAY) AS bigint
) AS window_start,
cast(max(ingest_time) AS BIGINT) - 8 * 3600 * 1000 AS current_ts,
count(DISTINCT id) AS part_daily_full_uv
FROM
view_table
GROUP BY
mod(id, 1024),
-- 根据注入时间划分天级别窗口
TUMBLE(ingest_time, INTERVAL '1' DAY)
--------------- 伪代码 ---------------
我目前所属的时区是东八区（北京时间），通过上述 sql，设置注入时间，并对注入时间加上 8 小时的偏移量进行天级别窗口的划分，就可以对此问题进行解决（也可以在 create table 时，在 schema 中根据计算列添加对应的注入时间戳进行解决）。如果你在 sql 层面有更好的解决方案，欢迎讨论~
Notes：
- 东 n 区的解决方案就是时间戳 +n * 3600 秒的偏移量，西 n 区的解决方案就是时间戳 -n * 3600 秒的偏移量
- DataStream API 存在相同的天级别窗口时区问题
这里提出一个问题，为什么东八区是需要在时间戳上加 8 小时偏移量进行天级别窗口计算，而不是减 8 小时或是加上 32（24 + 8） 小时，小伙伴们有详细分析过嘛~
根据上述问题，引出本文的第二大部分，即深度解析时区偏移量问题，这部分可以作为所有时区问题的分析思路。
4.为什么东八区是加 8 小时？
时间和时区基本概念
时区：由于世界各国家与地区经度不同，地方时也有所不同，因此会划分为不同的时区。
Unix 时间戳(Unix timestamp)： Unix 时间戳(Unix timestamp)，或称 Unix 时间(Unix time)、POSIX 时间(POSIX time)，是一种时间表示方式，定义为从格林威治时间 1970 年 01 月 01 日 00 时 00 分 00 秒（UTC/GMT的午夜）起至现在的总秒数。
Unix 时间戳不仅被使用在 Unix 系统、类 Unix 系统中，也在许多其他操作系统中被广泛采用。
GMT：Greenwich Mean Time 格林威治标准时间。这是以英国格林威治天文台观测结果得出的时间，这是英国格林威治当地时间，这个地方的当地时间过去被当成世界标准的时间。
UT：Universal Time 世界时。根据原子钟计算出来的时间。
UTC：Coordinated Universal Time 协调世界时。因为地球自转越来越慢，每年都会比前一年多出零点几秒，每隔几年协调世界时组织都会给世界时 +1 秒，让基于原子钟的世界时和基于天文学（人类感知）的格林威治标准时间相差不至于太大。并将得到的时间称为 UTC，这是现在使用的世界标准时间。
协调世界时不与任何地区位置相关，也不代表此刻某地的时间，所以在说明某地时间时要加上时区也就是说 GMT 并不等于 UTC，而是等于 UTC + 0，只是格林威治刚好在 0 时区上。
白话时间和时区
当时看完这一系列的时间以及时区说明之后我大脑其实是一片空白。...ojbk...，我用自己现在的一些理解，尝试将上述所有涉及到时间的概念解释一下。
- GMT：格林威治标准时间。
- UTC：基于原子钟协调之后的世界标准时间。可以认为 UTC 时间和格林威治标准时间一致。即 GMT = UTC + 0，其中 0 代表格林威治为 0 时区。
-
时区：逆向思维来解释下（只从技术层面解释，不从其他复杂层面解释），没有时区划分代表着全世界都是同一时区，那么同一时刻看到的外显时间是一样的。举个🌰：假如全世界都按照格林威治时间作为统一时间，在格林威治时间 0 点时，对于北京和加拿大的两个同学来说，这两个同学感知到的是北京是太阳刚刚升起（清晨），加拿大是太阳刚刚落下（傍晚）。
但是由于没有时区划分，这两个同学看到的时间都是 0 点，因此这是不符合人类对感知到的时间和自己看到的时间的理解的。所以划分时区之后，可以满足北京（东八区 UTC + 8）同学看到的时间是上午 8 点，加拿大（西四区 UTC - 4）同学看到的时间是下午 8 点。注意时区的划分是和 UTC 绑定的。东八区即 UTC + 8。 - flink 时间：flink 使用的时间基于 java 时间纪元（GMT 1970/01/01 00:00:00，UTC + 0 1970/01/01 00:00:00）。
- Unix 时间戳：世界上任何一个地方，同时接收到的数据的对应的 Unix 时间戳都是相同的，类似时区中我们举的不分时区的🌰，全世界同一时刻的 Unix 时间戳一致。
- Unix 时间戳为 0：对应的格林威治时间：1970-01-01 00:00:00，对应的北京时间（东八区）：1970-01-01 08:00:00**
概念关系如图所示：
time1.png
为什么东八区是加 8 小时？
下述表格只对一些重要的时间进行了标注：
| Unix 时间戳 | 格林威治时间（外显） | 北京时间（外显） |
|---|---|---|
| -8 * 3600 | - | 1970/01/01 00:00:00 |
| 0 | 1970/01/01 00:00:00 | 1970/01/01 08:00:00 |
| 16 * 3600 | - | 1970/01/02 00:00:00 |
| 24 * 3600 | 1970/01/02 00:00:00 | - |
time.png
拿第一条数据解释下，其代表在北京时间 1970/01/01 00:00:00 时，生成的一条数据所携带的 Unix 时间戳为 -8 * 3600。
根据需求和上图和上述表格内容，我们可以得到如下推导过程：
-
需求场景是统计一个整天的 uv，即天级别窗口，比如统计北京时间 1970/01/01 00:00:00 - 1970/01/02 00:00:00 范围的数据时，这个日期范围内的数据所携带的 Unix 时间戳范围为 -8 * 3600 到 16 * 3600
-
对于 flink 来说，默认情况下它所能统计的一个整天的 Unix 时间戳的范围是 0 到 24 * 3600
-
所以当我们想通过 flink 实现正确统计北京时间（1970/01/01 00:00:00 - 1970/01/02 00:00:00）范围内的数据时，即统计 Unix 时间戳为 -8 * 3600 到 16 * 3600 的数据时，就需要对时间戳做个映射。
-
映射方法如下，就是将整体范围内的时间戳做在时间轴上做平移映射，就是把 -8 * 3600 映射到 0，16 * 3600 映射到 24 * 3600。相当于是对北京时间的 Unix 时间戳整体加 8 * 3600。
-
最后在产出的时间戳上把加上的 8 小时再减掉（因为外显时间会自动按照时区对 Unix 时间戳进行格式化）。
Notes：
- 可以加 32 小时吗？答案是可以。在东八区，对于天级别窗口的划分，加 8 小时和加 8 + n * 24（其中 n 为整数）小时后进行的天级别窗口划分和计算的效果是一样的，flink 都会将东八区的整一天内的数据划分到一个天级别窗口内。所以加 32（8 + 24），56（8 + 48），-16（8 - 24）小时效果都相同，上述例子只是选择了时间轴平移最小的距离，即 8 小时。注意某些系统的 Unix 时间戳为负值时会出现异常。
- 此推理过程适用于所有遇到时区问题的场景，如果你也有其他应用场景有这个问题，也可以按照上述方式解决
Appendix
求输入 Unix 时间戳对应的东八区每天 0 点的 Unix 时间戳。
public static final long ONE_DAY_MILLS = 24 * 60 * 60 * 1000L;
public static long transform(long timestamp) {
return timestamp - (timestamp + 8 * 60 * 60 * 1000) % ONE_DAY_MILLS;
}
5.总结
本文首先介绍了直接给出了我们的问题 sql 和解决方案。
第二节从需求场景以及整个数据链路的实现方案出发，解释了我们怎样使用 flink sql 进行了需求实现，并进而引出了 sql 中天级别窗口存在的时区问题。
第三节确认了天级别窗口时区问题原因，引出了 flink 使用了 java 时间纪元，并针对此问题给出了引擎层面和 sql 层面的解决方案。也进而提出了一个问题：为什么我们的解决方案是加 8 小时偏移量？
第四节针对加 8 小时偏移量的原因进行了分析，并详细阐述了时区，UTC，GMT，Unix 时间戳之间的关系。
最后一节对本文进行了总结。
如果你有更方便的时区偏移量理解方式，欢迎留言~
