当前位置: 首页>编程语言>正文

flink支持的时间窗口 flink 时间窗口不固定

本系列每篇文章都是从一些实际的 case 出发,分析一些生产环境中经常会遇到的问题

本文主要分为两部分:

第一部分(第 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:数据源以及数据汇时区都为东八区)

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_数据分析,第1张

「没错,天级别窗口所存在的时区问题,即这段代码统计的不是楼主所在东八区一整天数据的 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 看板进行展示,没有任何维度字段(感动到哭????)。

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_flink支持的时间窗口_02,第2张

数据链路以及组件选型

客户端用户行为埋点日志 -> 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。

3.问题定位

问题说明

flink 在使用时间的这个概念的时候是基于 java 时间纪元(即格林威治 1970/01/01 00:00:00,也即 Unix 时间戳为 0)概念的,窗口对齐以及触发也是基于 java 时间纪元[1]

问题场景复现

可以通过直接查看 sink 数据的 window_start 得出上述结论。

但为了还原整个过程,我们按照如下 source 和 sink 数据进行整个问题的复现:

source 数据如下:

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_大数据_03,第3张

sink 数据(「为了方便理解,直接按照 druid 聚合之后的数据展示」):

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_数据分析_04,第4张

从上述数据可以发现,天级别窗口「开始时间」在 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 层面有更好的解决方案,欢迎讨论~

Notes:

  • 「东 n 区的解决方案就是时间戳 +n * 3600 秒的偏移量,西 n 区的解决方案就是时间戳 -n * 3600 秒的偏移量」
  • 「DataStream API 存在相同的天级别窗口时区问题」

这里提出一个问题,为什么东八区是需要在时间戳上加 8 小时偏移量进行天级别窗口计算,而不是减 8 小时或是加上 32(24 + 8) 小时,小伙伴们有详细分析过嘛~

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_编程语言_05,第5张

根据上述问题,引出本文的第二大部分,即深度解析时区偏移量问题,这部分可以作为所有时区问题的分析思路。

4.为什么东八区是加 8 小时?

时间和时区基本概念

「时区[3]:由于世界各国家与地区经度不同,地方时也有所不同,因此会划分为不同的时区。

「Unix 时间戳(Unix timestamp)[4]: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**

概念关系如图所示:

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_编程语言_06,第6张

为什么东八区是加 8 小时?

下述表格只对一些重要的时间进行了标注:

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_大数据_07,第7张

拿第一条数据解释下,其代表在北京时间 1970/01/01 00:00:00 时,生成的一条数据所携带的 Unix 时间戳为 -8 * 3600。

flink支持的时间窗口 flink 时间窗口不固定,flink支持的时间窗口 flink 时间窗口不固定_人工智能_08,第8张

根据需求和上图和上述表格内容,我们可以得到如下推导过程:

  • 需求场景是统计一个整天的 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 时间戳为负值时会出现异常。」
  • 「此推理过程适用于所有遇到时区问题的场景,如果你也有其他应用场景有这个问题,也可以按照上述方式解决」

附赠

求输入 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 时间戳之间的关系。

最后一节对本文进行了总结。

如果你有更方便的时区偏移量理解方式,欢迎留言~

Reference

[1]

java 时间纪元: https://cloud.tencent.com/developer/article/1447368


[2]

Blink Planner 时区设置: https://www.alibabacloud.com/help/zh/doc-detail/96910.htm?spm=a2c63.p38356.b99.48.28613830DXb6FQ


[3]

时区: https://baike.baidu.com/item/%E6%97%B6%E5%8C%BA/491122?fr=aladdin


[4]

Unix 时间戳(Unix timestamp): https://baike.baidu.com/item/unix%E6%97%B6%E9%97%B4%E6%88%B3/2078227?fr=aladdin


https://www.xamrdz.com/lan/5fg1941960.html

相关文章: