本系列每篇⽂章都是从⼀些实际的 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。
时区问题
❝
巴拉圭时间18：
「头⽂字 ∩ 技术⼩哥哥」：使⽤ sql，easy game，闲坐摸鱼...
「头⽂字 ∩ 技术⼩哥哥」：等到 「00:00」 时，发现指标还在不停地往上涨，难道是 sql 逻辑错了，不应该啊，试过分钟，⼩时级别窗⼝都⽊有这个问题
「头⽂字 ∩ 技术⼩哥哥」：抠头ing，算了，稍后再分析这个问题吧，现在还有正事要⼲
「头⽂字 ∩ 技术⼩哥哥」：到了早上，瞅了⼀眼配置的时间序列报表，发现在 「08:00」 点的时候指标归零，重新开始累计。
woc，想法⼀闪⽽过，东⼋区？（当时为啥没 format 下 sink 数据中的 ）
❞
3.问题定位
问题说明
flink 在使⽤时间的这个概念的时候是基于 java 时间纪元（即格林威治 1970/01/01 00:00:00，也即 Unix 时间戳为 0）概念的，窗⼝对齐以及触发也是基于 java 时间纪元[1]。
问题场景复现
可以通过直接查看 sink 数据的 window_start 得出上述结论。
但为了还原整个过程，我们按照如下 source 和 sink 数据进⾏整个问题的复现：
source 数据如下：
sink 数据（「为了⽅便理解，直接按照 druid 聚合之后的数据展⽰」）：
从上述数据可以发现，天级别窗⼝「开始时间」在 UTC + 8（北京）的时区是每天早上 8 点，即 UTC + 0（格林威治）的凌晨 0 点。「下⽂先给出解决⽅案，然后详细解析各个时间以及时区概念~」
解决⽅案
「框架层⾯解决」：Blink Planner ⽀持时区设置[2]
「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 层⾯有更好的解决⽅案，欢迎讨论~
这⾥提出⼀个问题，为什么东⼋区是需要在时间戳上加 8 ⼩时偏移量进⾏天级别窗⼝计算，⽽不是减 8 ⼩时或是加上 32（24 + 8） ⼩时，⼩伙伴们有详细分析过嘛~
