Flink SQL时间属性和窗口介绍

(1)概述

时间属性(time attributes),其实就是每个表模式结构(schema)的一部分。它可以在创建表的 DDL 里直接定义为一个字段,也可以在 DataStream 转换成表时定义。

一旦定义了时间属性,它就可以作为一个普通字段引用,并且可以在基于时间的操作中使用。

时间属性的数据类型为 TIMESTAMP,它的行为类似于常规时间戳,可以直接访问并且进行计算。

按照时间语义的不同,可以把时间属性的定义分成事件时间(event time)和处理时间(processing time)两种情况。

(2)事件时间

1)在创建表的 DDL 中定义

在创建表的 DDL(CREATE TABLE 语句)中,可以增加一个字段,通过 WATERMARK语句来定义事件时间属性。

WATERMARK 语句主要用来定义水位线(watermark)的生成表达式,这个表达式会将带有事件时间戳的字段标记为事件时间属性,并在它基础上给出水位线的延迟时间。

具体定义方式如下:

CREATE TABLE EventTable(user STRING,url STRING,ts TIMESTAMP(3),WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (...
);

把 ts 字段定义为事件时间属性,而且基于 ts 设置了 5 秒的水位线延迟。这里的“5 秒”是以“时间间隔”的形式定义的,格式是 INTERVAL <数值> <时间单位>:INTERVAL '5’ SECOND,这里的数值必须用单引号引起来,而单位用 SECOND 和 SECONDS 是等效的。

Flink 中支持的事件时间属性数据类型必须为 TIMESTAMP 或者 TIMESTAMP_LTZ。这里TIMESTAMP_LTZ 是指带有本地时区信息的时间戳(TIMESTAMP WITH LOCAL TIME ZONE);

如数据中的时间戳是“年-月-日-时-分-秒”形式,那就是不带时区信息的,可以将事件时间属性定义为 TIMESTAMP 类型。
而如果原始的时间戳就是一个长整型的毫秒数,这时就需要另外定义一个字段来表示事件时间属性,类型定义为 TIMESTAMP_LTZ 会更方便:

CREATE TABLE events (user STRING,url STRING,ts BIGINT,ts_ltz AS TO_TIMESTAMP_LTZ(ts, 3),WATERMARK FOR ts_ltz AS ts_ltz - INTERVAL '5' SECOND
) WITH (...
);

这里我们另外定义了一个字段 ts_ltz,是把长整型的 ts 转换为 TIMESTAMP_LTZ 得到的;进而使用 WATERMARK 语句将它设为事件时间属性,并设置 5 秒的水位线延迟。

2)在数据流转换为表时定义

调用 fromDataStream()方法创建表时,可以追加参数来定义表中的字段结构;这时可以给某个字段加上.rowtime() 后缀,就表示将当前字段指定为事件时间属性。

这个字段可以是数据中本不存在、额外追加上去的“逻辑字段”,就像之前 DDL 中定义的第二种情况;也可以是本身固有的字段,那么这个字段就会被事件时间属性所覆盖,类型也会被转换为 TIMESTAMP。

不论哪种方式,时间属性字段中保存的都是事件的时间戳(TIMESTAMP 类型)。

需要注意的是,这种方式只负责指定时间属性,而时间戳的提取和水位线的生成应该之前就在 DataStream 上定义好了。

由于 DataStream 中没有时区概念,因此 Flink 会将事件时间属性解析成不带时区的 TIMESTAMP 类型,所有的时间值都被当作 UTC 标准时间。

在代码中的定义方式如下:

// 方法一:
// 流中数据类型为二元组 Tuple2,包含两个字段;需要自定义提取时间戳并生成水位线
DataStream<Tuple2<String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);// 声明一个额外的逻辑字段作为事件时间属性
Table table = tEnv.fromDataStream(stream, $("user"), $("url"),$("ts").rowtime());// 方法二:
// 流中数据类型为三元组 Tuple3,最后一个字段就是事件时间戳
DataStream<Tuple3<String, String, Long>> stream = inputStream.assignTimestampsAndWatermarks(...);// 不再声明额外字段,直接用最后一个字段作为事件时间属性
Table table = tEnv.fromDataStream(stream, $("user"), $("url"),$("ts").rowtime());

(3)处理时间

系统时间,使用时不需要提取时间戳(timestamp)和生成水位线(watermark)。因此在定义处理时间属性时,必须要额外声明一个字段,专门用来保存当前的处理时间。

类似地,处理时间属性的定义也有两种方式:创建表 DDL 中定义,或者在数据流转换成表时定义。

1)在创建表的 DDL 中定义

在创建表的 DDL(CREATE TABLE 语句)中,可以增加一个额外的字段,通过调用系统内置的 PROCTIME()函数来指定当前的处理时间属性,返回的类型是 TIMESTAMP_LTZ。

CREATE TABLE EventTable(user STRING,url STRING,ts AS PROCTIME()
) WITH (...
);

时间属性,以“计算列”(computed column)的形式定义出来的。所谓的计算列是 Flink SQL 中引入的特殊概念,可以用一个 AS 语句来在表中产生数据中不存在的列,并且可以利用原有的列、各种运算符及内置函数。

在前面事件时间属性的定义中,将 ts 字段转换成 TIMESTAMP_LTZ 类型的 ts_ltz,也是计算列的定义方式。

2)在数据流转换为表时定义

处理时间属性同样可以在将 DataStream 转换为表的时候来定义。我们调用 fromDataStream()方法创建表时,可以用.proctime()后缀来指定处理时间属性字段。

由于处理时间是系统时间,原始数据中并没有这个字段,所以处理时间属性一定不能定义在一个已有字段上,只能定义在表结构所有字段的最后,作为额外的逻辑字段出现。

代码中定义处理时间属性的方法如下:

DataStream<Tuple2<String, String>> stream = ...;// 声明一个额外的字段作为处理时间属性字段
Table table = tEnv.fromDataStream(stream, $("user"), $("url"),$("ts").proctime());

(4)窗口(Window)

1)分组窗口(Group Window,老版本)

在 Flink 1.12 之前的版本中,Table API 和 SQL 提供了一组“分组窗口”(Group Window)函数,常用的时间窗口如滚动窗口、滑动窗口、会话窗口都有对应的实现;具体在 SQL 中就是调用 TUMBLE()、HOP()、SESSION(),传入时间属性字段、窗口大小等参数就可以了。

以滚动窗口为例:

TUMBLE(ts, INTERVAL '1' HOUR)

这里的 ts 是定义好的时间属性字段,窗口大小用“时间间隔”INTERVAL 来定义。在进行窗口计算时,分组窗口是将窗口本身当作一个字段对数据进行分组的,可以对组内的数据进行聚合。

基本使用方式如下:

Table result = tableEnv.sqlQuery("SELECT " +"user, " +
"TUMBLE_END(ts, INTERVAL '1' HOUR) as endT, " +"COUNT(url) AS cnt " +"FROM EventTable " +"GROUP BY " + // 使用窗口和用户名进行分组"user, " +"TUMBLE(ts, INTERVAL '1' HOUR)" // 定义 1 小时滚动窗口);

这里定义了 1 小时的滚动窗口,将窗口和用户 user 一起作为分组的字段。用聚合函数COUNT()对分组数据的个数进行了聚合统计,并将结果字段重命名为cnt;用TUPMBLE_END() 函数获取滚动窗口的结束时间,重命名为 endT 提取出来。

分组窗口的功能比较有限,只支持窗口聚合,所以目前已经处于弃用(deprecated)的状态。

2)窗口表值函数(Windowing TVFs,新版本)

从 1.13 版本开始,Flink 开始使用窗口表值函数(Windowing table-valued functions,Windowing TVFs)来定义窗口。

窗口表值函数是 Flink 定义的多态表函数(PTF),可以将表进行扩展后返回。表函数(table function)可以看作是返回一个表的函数。

目前 Flink 提供了以下几个窗口 TVF:
1.滚动窗口(Tumbling Windows);
2.滑动窗口(Hop Windows,跳跃窗口);
3.累积窗口(Cumulate Windows);
4.会话窗口(Session Windows,目前尚未完全支持)。

1.概述

窗口表值函数可以完全替代传统的分组窗口函数。窗口 TVF 更符合 SQL 标准,性能得到了优化,拥有更强大的功能;可以支持基于窗口的复杂计算,例如窗口 Top-N、窗口联结(window join)等等。

在窗口 TVF 的返回值中,除去原始表中的所有列,还增加了用来描述窗口的额外 3 个列:“窗口起始点(window_start)、“窗口结束点”(window_end)、“窗口时间”(window_time)。

起始点和结束点比较好理解,这里的“窗口时间”指的是窗口中的时间属性,它的值等于 window_end - 1ms,所以相当于是窗口中能够包含数据的最大时间戳。

在 SQL 中的声明方式,与以前的分组窗口是类似的,直接调用 TUMBLE()、HOP()、CUMULATE()就可以实现滚动、滑动和累积窗口,不过传入的参数会有所不同。

2.滚动窗口(TUMBLE)

滚动窗口在 SQL 中的概念与 DataStream API 中的定义完全一样,是长度固定、时间对齐、无重叠的窗口,一般用于周期性的统计计算。

在 SQL 中通过调用 TUMBLE()函数就可以声明一个滚动窗口,只有一个核心参数就是窗口大小(size)。

在 SQL 中不考虑计数窗口,所以滚动窗口就是滚动时间窗口,参数中还需要将当前的时间属性字段传入;另外,窗口 TVF 本质上是表函数,可以对表进行扩展,所以还应该把当前查询的表作为参数整体传入。具体声明如下:

TUMBLE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR)

这里基于时间字段 ts,对表 EventTable 中的数据开了大小为 1 小时的滚动窗口。窗口会将表中的每一行数据,按照它们 ts 的值分配到一个指定的窗口中。

(2)滑动窗口(HOP)

滑动窗口的使用与滚动窗口类似,可以通过设置滑动步长来控制统计输出的频率。在 SQL中通过调用 HOP()来声明滑动窗口;除了也要传入表名、时间属性外,还需要传入窗口大小(size)和滑动步长(slide)两个参数。

HOP(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '1' HOURS));

这里我们基于时间属性 ts,在表 EventTable 上创建了大小为 1 小时的滑动窗口,每 5 分钟滑动一次。需要注意的是,紧跟在时间属性字段后面的第三个参数是步长(slide),第四个参数才是窗口大小(size)。

(3)累积窗口(CUMULATE)

滚动窗口和滑动窗口,可以用来计算大多数周期性的统计指标。

在实际应用中还会遇到这样一类需求:我们的统计周期可能较长,因此希望中间每隔一段时间就输出一次当前的统计值;

与滑动窗口不同的是,在一个统计周期内,我们会多次输出统计值,它们应该是不断叠加累积的。

例如,我们按天来统计网站的 PV(Page View,页面浏览量),如果用 1 天的滚动窗口,那需要到每天 24 点才会计算一次,输出频率太低;如果用滑动窗口,计算频率可以更高,但统计的就变成了“过去 24 小时的 PV”。

所以我们真正希望的是,还是按照自然日统计每天的PV,不过需要每隔 1 小时就输出一次当天到目前为止的 PV 值。这种特殊的窗口就叫作“累积窗口”(Cumulate Window)。

在这里插入图片描述

累积窗口是窗口 TVF 中新增的窗口功能,它会在一定的统计周期内进行累积计算。

累积窗口中有两个核心的参数:最大窗口长度(max window size)和累积步长(step)。

所谓的最大窗口长度其实就是我们所说的“统计周期”,最终目的就是统计这段时间内的数据。开始时,创建的第一个窗口大小就是步长 step;之后的每个窗口都会在之前的基础上再扩展 step 的长度,直到达到最大窗口长度。

在 SQL 中可以用 CUMULATE()函数来定义,具体如下:

CUMULATE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOURS, INTERVAL '1' DAYS))

这里我们基于时间属性 ts,在表 EventTable 上定义了一个统计周期为 1 天、累积步长为 1小时的累积窗口。注意第三个参数为步长 step,第四个参数则是最大窗口长度。

上面所有的语句只是定义了窗口,类似于 DataStream API 中的窗口分配器;在 SQL 中窗口的完整调用,还需要配合聚合操作和其它操作。

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

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

相关文章

数票通批量开票小妙招,双11开票不用愁

每年双十一都是电商企业和财务“疯狂”的日子&#xff0c;随着疫情的回暖&#xff0c;消费者的消费能力蹭蹭上涨&#xff0c;商家的口袋盆满钵满&#xff0c;而财务的加班永无休止&#xff0c;批量开票成为财务人摆脱双十一加班的必备技能。 面对双十一狂欢&#xff0c;相信不少…

高校双十一快递爆仓警告,做快递代取能不能赚钱

今年双11期间的快递数量预计将超过往年。今年以来&#xff0c;快递行业竞争日益激烈&#xff0c;特别是价格战的竞争更是白热化。电商平台也针对低价展开了一系列促销活动&#xff0c;这种策略有望刺激快递单量增长。低价吸引力激发了用户的购物欲望&#xff0c;将导致高校快递…

【数据结构】树与二叉树(二):树的表示C语言:树形表示法、嵌套集合表示法、嵌套括号表示法 、凹入表示法

文章目录 5.1 树的基本概念5.1.1 树的定义5.1.2 森林的定义5.1.3 树的术语5.1.4 树的表示1&#xff0e;树形表示法2&#xff0e;嵌套集合表示法结构体创建树主函数 3&#xff0e;嵌套括号表示法结构体创建树嵌套括号表示法主函数 4&#xff0e;凹入表示法结构体创建树凹入表示法…

【Mybatis小白从0到90%精讲】15: Mybatis配置打印SQL日志

文章目录 前言配置日志实现前言 日志(Log)是每个程序都不可或缺的一部分,它可以帮助开发人员诊断和调试问题。Mybatis,作为一款备受赞誉的ORM框架,自然也提供了强大的日志功能。 它不仅提供了内置的标准实现,还支持集成各种主流的日志框架,让我们可以轻松地查看最终执行…

C-DS二叉树_另一棵树的子树

Description 给你两棵二叉树tree1和tree2,检验tree1中是否包含和tree2具有相同结构和结点值的子树。如果存在,输出true;否则,输出false。 Input 第一行输入t,表示有t个测试样例。 第二行首先输入n1,接着输入n1个整数,表示二叉树tree1。 第三行首先输入n2,接着输入n…

常见面试题-MySQL专栏(三)MVCC、BufferPool

typora-copy-images-to: imgs 了解 MVCC 吗&#xff1f; 答&#xff1a; MVCC&#xff08;Multi-Version Concurrency Control&#xff09; 是用来保证 MySQL 的事务隔离性的&#xff0c;对一行数据的读和写两个操作默认是不会通过加锁互斥来保证隔离性&#xff0c;避免了频…

2023 现阶段H5的机型适配

个人愚见 现在的主流体验&#xff0c;是大屏手机展示更多的内容&#xff0c;并不着重于放大展示&#xff0c; 所以&#xff0c;外层布局使用vw,百分比&#xff0c;flex&#xff0c;内层直接px就行 比如微信服务页面&#xff0c;大屏下展示更多数据

晨控CK-GW08系列网关控制器与CODESYS软件MODBUSTCP通讯手册

晨控CK-GW08系列是一款支持标准工业通讯协议ModbusTCP的网关控制器,方便用户集成到PLC等控制系统中。系统还集成了8路读写接口&#xff0c;用户可通过通信接口使用Modbus TCP协议对8路读写接口所连接的读卡器进行相对独立的读写操作。 晨控CK-GW08系列网关控制器适用于本公司多…

Flink往Starrocks写数据报错:too many filtered rows

Bug信息 Caused by: com.starrocks.data.load.stream.exception.StreamLoadFailException: {"TxnId": 2711690,"Label": "cd528707-8595-4a35-b2bc-39b21087d6ec","Status": "Fail","Message": "too many f…

Git 安全警告修复手册:解决 `fatal: detected dubious ownership in repository at ` 问题 ️

&#x1f337;&#x1f341; 博主猫头虎 带您 Go to New World.✨&#x1f341; &#x1f984; 博客首页——猫头虎的博客&#x1f390; &#x1f433;《面试题大全专栏》 文章图文并茂&#x1f995;生动形象&#x1f996;简单易学&#xff01;欢迎大家来踩踩~&#x1f33a; &a…

httpclient工具类(支持泛型转换)

1、网上搜到的httpclient工具类的问题&#xff1a; 1.1、如下图我们都能够发现这种封装的问题&#xff1a; 代码繁杂、充斥了很多重复性代码返回值单一&#xff0c;无法拿到对应的Java Bean对象及List对象集合实际场景中会对接大量第三方的OPEN API&#xff0c;下述方法的扩展…

【电路笔记】-谐波

谐波 文章目录 谐波1、概述2、频谱分析3、已知信号4、未知信号5、总结 周期性信号并不总是完美的正弦模式&#xff0c;例如我们之前有关 正弦波的文章之一中介绍的那样。 有时&#xff0c;信号确实可以是简单正弦波的叠加&#xff0c;它们被称为复杂波形。 在本文中&#xff0…