Flink:Temporal Table Function(时态表函数)和 Temporal Join

2024-03-03 14:04

本文主要是介绍Flink:Temporal Table Function(时态表函数)和 Temporal Join,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

《大数据平台架构与原型实现:数据中台建设实战》博主历时三年精心创作的《大数据平台架构与原型实现:数据中台建设实战》一书现已由知名IT图书品牌电子工业出版社博文视点出版发行,点击《重磅推荐:建大数据平台太难了!给我发个工程原型吧!》了解图书详情,京东购书链接:https://item.jd.com/12677623.html,扫描左侧二维码进入京东手机购书页面。

我们知道,时态表(确切地说应该是版本表)提供了回溯历史的能力,也就是能读取一条记录过去某个时刻所对应的值。要想查询版本表在过去某个时刻对应的值,我们得在查询时把这个时间作为参数传递给版本表,但这个时间参数绝不会是一个 where 条件,它是另一个维度(时间维度)上的参数,那么用怎样的形式才能把这个时间参数合理地表达到查询中呢? Flink 使用了 UDF 的形式,主要思路就是:注册一个 UDF 来指代一张版本表,表名不能有参数,但函数可以有,这时把想访问版本表的目标时间点作为参数传给这个UDF,返回的就是当时表中的数据了,这个 UDF 就被称作:Temporal Table Function!

例如:以下代码将汇率表 currency_rates 注册成了时态表函数 rates。(注意:目前在 Flink SQL 中是不支持定义 Temporal Table Function 的!只能以代码方式定义,但是 SQL 中可以定义 Temporal Table DDL

rates = tEnv.from("currency_rates").createTemporalTableFunction("update_time", "currency")tEnv.createTemporarySystemFunction("rates", rates); 

然后,使用下面的 SQL 就能查询出在 11:05 时的汇率信息了:

SELECT * FROM rates('11:05');

可以说:是时态表函数是访问时态表的“入口”,是时态表的“正确打开方式”!

但是,像上面那样直接查询某一时刻版本表上的数据的情形其实并不多,真正常见是:其他表主动 Join 一张时态表,期望获得表中记录所代表的事件在发生时刻时态表中的当时的数据,就是我们曾经解释的“当时对当时”的需求场景(典型案例:Join 汇率表计算订单当时的总价):

-- 基于时态表函数实现的Join,由于指定的 order_time 是一个事件时间
-- 所以该SQL实现的是:基于事件时间的 Temporal Join,也就是 Join 事件发生时刻关联表当时的值
SELECTSUM(amount * rate) AS amount
FROMorders,LATERAL TABLE (rates(order_time))
WHERErates.currency = orders.currency

上面的 SQL 就是标准的 Temporal Table Function Join 语法,SQL 中使用了关键字 LATERAL TABLE,填入一个 Temporal Table Function / 时态表函数 rates,设定传给时态表的时间属性(基于什么时间查找时态表上的版本)order_time

这里,官方文档其实隐去了一个背景信息,order_time 其实是 orders 表的事件时间属性,所以,上述使用 Temporal Table Function Join 语法实现的是:基于事件时间的 Temporal Join,这种 Join 还可以通过 FOR SYSTEM_TIME AS OF 关键字实现, Temporal Table Function Join 语法除了能实现基于事件时间的 Temporal Join 外,还能实现基于处理时间的 Temporal Join 了,语法不变,只要将传给 rates 函数的时间属性从一个事件时间改为一个处理时间就可以了,就像 [ 官方文档 ] 给出的示例中那样,使用了一个 o_proctime 字段,这个字段是 orders 表的处理时间属性:

-- 基于时态表函数实现的Join,由于指定的 o_proctime 是一个处理时间
-- 所以该SQL实现的是:基于处理时间的 Temporal Join,也就是总是 Join 关联表当前最新状态的数据
SELECTo_amount, r_rate
FROMOrders,LATERAL TABLE (rates(o_proctime))
WHEREr_currency = o_currency

这篇关于Flink:Temporal Table Function(时态表函数)和 Temporal Join的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/769838

相关文章

C++统计函数执行时间的最佳实践

《C++统计函数执行时间的最佳实践》在软件开发过程中,性能分析是优化程序的重要环节,了解函数的执行时间分布对于识别性能瓶颈至关重要,本文将分享一个C++函数执行时间统计工具,希望对大家有所帮助... 目录前言工具特性核心设计1. 数据结构设计2. 单例模式管理器3. RAII自动计时使用方法基本用法高级用法

GO语言中函数命名返回值的使用

《GO语言中函数命名返回值的使用》在Go语言中,函数可以为其返回值指定名称,这被称为命名返回值或命名返回参数,这种特性可以使代码更清晰,特别是在返回多个值时,感兴趣的可以了解一下... 目录基本语法函数命名返回特点代码示例命名特点基本语法func functionName(parameters) (nam

Python Counter 函数使用案例

《PythonCounter函数使用案例》Counter是collections模块中的一个类,专门用于对可迭代对象中的元素进行计数,接下来通过本文给大家介绍PythonCounter函数使用案例... 目录一、Counter函数概述二、基本使用案例(一)列表元素计数(二)字符串字符计数(三)元组计数三、C

Python中的filter() 函数的工作原理及应用技巧

《Python中的filter()函数的工作原理及应用技巧》Python的filter()函数用于筛选序列元素,返回迭代器,适合函数式编程,相比列表推导式,内存更优,尤其适用于大数据集,结合lamb... 目录前言一、基本概念基本语法二、使用方式1. 使用 lambda 函数2. 使用普通函数3. 使用 N

MySQL中REPLACE函数与语句举例详解

《MySQL中REPLACE函数与语句举例详解》在MySQL中REPLACE函数是一个用于处理字符串的强大工具,它的主要功能是替换字符串中的某些子字符串,:本文主要介绍MySQL中REPLACE函... 目录一、REPLACE()函数语法:参数说明:功能说明:示例:二、REPLACE INTO语句语法:参数

python中update()函数的用法和一些例子

《python中update()函数的用法和一些例子》update()方法是字典对象的方法,用于将一个字典中的键值对更新到另一个字典中,:本文主要介绍python中update()函数的用法和一些... 目录前言用法注意事项示例示例 1: 使用另一个字典来更新示例 2: 使用可迭代对象来更新示例 3: 使用

Python lambda函数(匿名函数)、参数类型与递归全解析

《Pythonlambda函数(匿名函数)、参数类型与递归全解析》本文详解Python中lambda匿名函数、灵活参数类型和递归函数三大进阶特性,分别介绍其定义、应用场景及注意事项,助力编写简洁高效... 目录一、lambda 匿名函数:简洁的单行函数1. lambda 的定义与基本用法2. lambda

Python 函数详解:从基础语法到高级使用技巧

《Python函数详解:从基础语法到高级使用技巧》本文基于实例代码,全面讲解Python函数的定义、参数传递、变量作用域及类型标注等知识点,帮助初学者快速掌握函数的使用技巧,感兴趣的朋友跟随小编一起... 目录一、函数的基本概念与作用二、函数的定义与调用1. 无参函数2. 带参函数3. 带返回值的函数4.

MySQL中DATE_FORMAT时间函数的使用小结

《MySQL中DATE_FORMAT时间函数的使用小结》本文主要介绍了MySQL中DATE_FORMAT时间函数的使用小结,用于格式化日期/时间字段,可提取年月、统计月份数据、精确到天,对大家的学习或... 目录前言DATE_FORMAT时间函数总结前言mysql可以使用DATE_FORMAT获取日期字段

Django中的函数视图和类视图以及路由的定义方式

《Django中的函数视图和类视图以及路由的定义方式》Django视图分函数视图和类视图,前者用函数处理请求,后者继承View类定义方法,路由使用path()、re_path()或url(),通过in... 目录函数视图类视图路由总路由函数视图的路由类视图定义路由总结Django允许接收的请求方法http