【翻译】Flink Table Api & SQL —Streaming 概念 ——在持续查询中 Join

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

本文翻译自官网： Joins in Continuous Queries https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streaming/joins.html

Flink Table Api & SQL 翻译目录

Join 是批量数据处理中连接两个关系行的常见且易于理解的操作。但是，动态表上的 join 语义不那么明显，甚至令人困惑。

因此，有一些方法可以使 Table API或SQL实际执行 join 。

有关语法的更多信息，请检查Table API和SQL中的 join 部分。

常规 join （Regular Joins）
时间窗口 join
与时态表函数 join
- 用法
- 处理时间时态（表函数） join
- 事件时间时态（表函数） join
时态表 join

用法

Regular Joins

常规 join 是最通用的 join 类型，在该 join 中，任何新记录的更改对 join 输入两侧都是可见的，并且会影响整个 join 结果。例如，如果左侧有一个新记录，则它将与右侧的所有以前和将来的记录合并在一起。

SELECT * FROM Orders
INNER JOIN Product
ON Orders.productId = Product.id

这些语义允许进行任何类型的更新（插入，更新，删除）的输入表。

但是，此操作有一个重要的含义：它要求将 join 输入的两端始终保持在Flink的状态。因此，如果一个或两个输入表持续增长，资源使用也将无限期增长。

Time-windowed Joins

时间窗口 join 由 join 谓词定义，该 join 谓词检查输入记录的时间属性是否在某些时间限制（即时间窗口）内。

SELECT *
FROM
Orders o,
Shipments s
WHERE o.id = s.orderId AND
o.ordertime BETWEEN s.shiptime - INTERVAL '4' HOUR AND s.shiptime

与常规 join 操作相比，这种 join 仅支持具有时间属性的 append-only 表。由于时间属性是准单调递增的，因此Flink可以从其状态中删除旧值，而不会影响结果的正确性。

Join with a Temporal Table Function

具有时态表函数的 join 将 append-only 表（左侧输入/探针侧，注：输入流）与临时表（右侧输入/构建侧，注：维表）join，即随时间变化并跟踪其变化的表（维表）。请查看相应的页面以获取有关时态表的更多信息。

以下示例显示了 append-only 表 Orders，该表与不断变化的货币汇率表 RatesHistory 结合在一起。

Orders是一个 append-only 表，代表给定 amount 和给定货币（currency）的付款。例如，在10:15 有一笔金额为 2 欧元的订单。

SELECT * FROM Orders;rowtime amount currency
======= ====== =========
10:15 2 Euro
10:30 1 US Dollar
10:32 50 Yen
10:52 3 Euro
11:04 5 US Dollar

RatesHistory表示日元汇率（汇率为1）不断变化的 append-only 表。例如，从 09:00 到 10:45 欧元对日元的汇率为 114。从10:45到11:15为116。

SELECT * FROM RatesHistory;rowtime currency rate
======= ======== ======
09:00 US Dollar 102
09:00 Euro 114
09:00 Yen 1
10:45 Euro 116
11:15 Euro 119
11:49 Pounds 108

假设我们要计算所有订单转换为通用货币（日元）的金额。

例如，我们要使用给定 rowtime（114）的适当转换率转换以下订单。

rowtime amount currency
======= ====== =========
10:15 2 Euro

如果不使用时态表的概念，则需要编写如下查询：

SELECT
SUM(o.amount * r.rate) AS amount
FROM Orders AS o,
RatesHistory AS r
WHERE r.currency = o.currency
AND r.rowtime = (
SELECT MAX(rowtime)
FROM RatesHistory AS r2
WHERE r2.currency = o.currency
AND r2.rowtime <= o.rowtime);

借助时态表函数使 RatesHistory 的汇率变化，我们可以在SQL中将查询表示为：

SELECT
o.amount * r.rate AS amount
FROM
Orders AS o,
LATERAL TABLE (Rates(o.rowtime)) AS r
WHERE r.currency = o.currency

探针端记录的相关时间属性时，来自探针端的每个记录将与构建端表的版本关联。为了支持生成侧表上先前值的更新（覆盖），该表必须定义一个主键。

在我们的示例中，from Orders 中的每个记录都将与Rates 的 o.rowtime 时间版本结合在一起。该 currency 字段已提前定义为 Rates 的主键，并且在我们的示例中用于连接两个表。如果查询使用的是处理时间概念，则执行操作时，新添加的订单将始终与 Rates 的最新版本结合在一起。

与常规 join相反，这意味着如果在构建端有新记录，则不会影响 join的先前结果。这又使Flink可以限制必须保持状态的元素数量。

与时间窗口连接相比，时态表 join 未定义时间范围，（所有）关联的数据将被 join。探测端的记录总是在 time 属性指定的时间与构建端的版本连接在一起。因此，构建端（时态表）的记录可能是任意旧的。随着时间的流逝，该记录的先前版本和不再需要的版本（对于给定的主键）将从状态中删除。

这种行为使临时表成为一个很好的候选者，可以用关系术语来表示流的 join。

用法

定义时态表函数后，我们就可以开始使用它。可以使用与普通表函数相同的方式来使用时态表函数。

以下代码段解决了我们从Orders表中转换货币的初衷问题：

## SQL
SELECT
SUM(o_amount * r_rate) AS amount
FROM
Orders,
LATERAL TABLE (Rates(o_proctime))
WHERE
r_currency = o_currency## Java
Table result = orders
.joinLateral("rates(o_proctime)", "o_currency = r_currency")
.select("(o_amount * r_rate).sum as amount");## Scala
val result = orders
.joinLateral(rates('o_proctime), 'r_currency === 'o_currency)
.select(('o_amount * 'r_rate).sum as 'amount)

注意：对于时态表 join ，尚未实现在查询配置中定义的状态保留。这意味着，计算查询结果所需的状态可能会无限增长，具体取决于历史记录表的不同主键数量。

Processing-time Temporal Joins

基于处理时间的时间属性，是不可能通过 过去的 时间的属性作为参数的时间表函数（注：处理时间只会增长）。根据定义，它始终是当前时间戳。因此，处理时间时态表函数的调用将始终返回基础表的最新已知版本，基础历史表中的任何更新也将立即覆盖当前值。

仅将构建侧记录的最新版本（相对于定义的主键）保持在该状态。构建端的更新将不会影响先前发出的 join 结果。

可以将处理时间的时态表视为一种简单的 HashMap<K, V>，它可以存储构建侧的所有记录。当来自构建端的新记录与先前的记录具有相同的键时，旧值将被简单地覆盖。总是根据HashMap的最新/当前状态评估来自探测器端的每个记录（注：与输入数据 join）。

Event-time Temporal Joins

使用事件时间的时间属性（即行时间属性），可以将过去的时间属性传递给时态表函数。这允许在公共时间点将两个表（时态表的两个时间状态）连接在一起。

与处理时间时态表 join 相比，时态表不仅将构建侧记录的最新版本（相对于定义的主键）保持在状态中，而且还存储自上次水印以来的所有版本（按时间标识）。

例如，根据时态表的概念，将附加到探针侧表的事件时间时间戳为12:30:00的输入行与构建侧表在时间12:30:00的版本进行连接。因此，传入行仅与时间戳小于或等于12:30:00的行连接，并根据主键应用更新直到该时间点。

根据事件时间的定义，水印允许 join 操作及时向前移动，并丢弃不再需要的构建表版本，因为不会输入具有更低或相等时间戳的行。

与时态表 Join

与时态表的 join 将任意表（左侧输入/探针侧）与时态表（右侧输入/构建侧）join，即随时间变化的外部数据表。请查看相应的页面以获取有关时态表的更多信息。

注意：用户不能将任意表用作时态表，需要使用 LookupableTableSource 支持的表。LookupableTableSource 只能作为时间表用于时间联接。有关如何定义LookupableTableSource的更多详细信息，请参见页面。

下面的示例显示了一个Orders流，该流与不断变化的货币汇率表 LatestRates 结合在一起。

LatestRates是物化最新汇率的维度表。在时间10:15、10:30、10:52，LatestRates 的内容如下：

10:15> SELECT * FROM LatestRates;currency rate
======== ======
US Dollar 102
Euro 114
Yen 110:30> SELECT * FROM LatestRates;currency rate
======== ======
US Dollar 102
Euro 114
Yen 110:52> SELECT * FROM LatestRates;currency rate
======== ======
US Dollar 102
Euro 116 <==== changed from 114 to 116
Yen 1

时间10:15和10:30的LastestRates的内容相等。欧元汇率在10:52从114更改为116。

Orders 是一个 append-only 表，代表给定金额和给定货币的付款。例如，在10:15时有一笔2欧元的订单。

SELECT * FROM Orders;amount currency
====== =========
2 Euro <== arrived at time 10:15
1 US Dollar <== arrived at time 10:30
2 Euro <== arrived at time 10:52

假设我们要计算所有Orders折算成通用货币（日元）的金额。

例如，我们想使用LatestRates中的最新汇率转换以下订单。结果将是：

amount currency rate amout*rate
====== ========= ======= ============
2 Euro 114 228 <== arrived at time 10:15
1 US Dollar 102 102 <== arrived at time 10:30
2 Euro 116 232 <== arrived at time 10:52

借助时态表联接，我们可以在SQL中将查询表示为：

SELECT
o.amout, o.currency, r.rate, o.amount * r.rate
FROM
Orders AS o
JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r
ON r.currency = o.currency

探针端的每个记录都将与构建端表的当前版本关联。在我们的示例中，查询使用的是处理时间概念，因此在执行操作时，新附加的订单将始终与最新版本的LatestRates结合在一起。注意，结果对于处理时间不是确定的。

与常规 join相反，尽管在构建方面进行了更改（注：数据修改了），但时态表 Join 的先前结果将不会受到影响。而且，时态表 join 运算符非常轻，并且不保留任何状态。

与时间窗口 join相比，时态表 join 没有定义将要在其中 join 记录的时间窗口。在处理时，探测端的记录总是与构建端的最新版本结合在一起。因此，构建方面的记录可能是任意旧的。

时态表函数 join 和时态表 join 都来自相同的初衷，但是具有不同的SQL语法和运行时实现：

时态表函数 join 的SQL语法是 join UDTF，而时态表联接使用 SQL：2011 中引入的标准时态表语法。
时态表函数 join 的实现实际上 join 了两个流并使它们保持状态，而时态表 join 仅接收唯一的输入流并根据记录中的键查找外部数据库。
时态表函数 join 通常用于 join 变更日志流，而临时表 join 通常用于 join 外部表（即维表）。

这种行为使时态表成为一个很好的候选者，可以用关系术语来表示流的 join。

将来，时态表联接将支持时态表功能 join 的功能，即支持时态 join 变更日志流。

用法

临时表 join 的语法如下：

SELECT [column_list]
FROM table1 [AS <alias1>]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.proctime [AS <alias2>]
ON table1.column-name1 = table2.column-name1

当前，仅支持 INNER JOIN 和 LEFT JOIN。临时表之后应遵循 FOR SYSTEM_TIME AS OF table1.proctime .proctime是table1的处理时间属性。这意味着在连接左表中的每个记录时，它会在处理时为时态表做快照。

例如，在定义时态表之后，我们可以如下使用它。

SELECT
SUM(o_amount * r_rate) AS amount
FROM
Orders
JOIN LatestRates FOR SYSTEM_TIME AS OF o_proctime
ON r_currency = o_currency

注意：仅在 Blink planner 中支持。

注意：仅在SQL中支持，而在 Table API 中尚不支持。

注意：Flink当前不支持事件时间时态表 join。

欢迎关注Flink菜鸟公众号，会不定期更新Flink（开发技术）相关的推文

【翻译】Flink Table Api & SQL —Streaming 概念 ——在持续查询中 Join的更多相关文章

【翻译】Flink Table Api &amp&semi; SQL —Streaming 概念 ——动态表
本文翻译自官网:Flink Table Api & SQL 动态表 https://ci.apache.org/projects/flink/flink-docs-release-1.9/de ...
【翻译】Flink Table Api &amp&semi; SQL —Streaming 概念 —— 时态表
本文翻译自官网: Temporal Tables https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/strea ...
【翻译】Flink Table Api &amp&semi; SQL ——Streaming 概念
本文翻译自官网:Streaming 概念 https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/streamin ...
【翻译】Flink Table Api &amp&semi; SQL —Streaming 概念 ——时间属性
本文翻译自官网: Time Attributes https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/str ...
【翻译】Flink Table Api &amp&semi; SQL —Streaming 概念 —— 表中的模式匹配 Beta版
本文翻译自官网:Detecting Patterns in Tables Beta https://ci.apache.org/projects/flink/flink-docs-release-1 ...
【翻译】Flink Table Api &amp&semi; SQL —Streaming 概念 —— 查询配置
本文翻译自官网:Query Configuration https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/s ...
【翻译】Flink Table Api &amp&semi; SQL — 流概念
本文翻译自官网:Streaming Concepts https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/st ...
Flink Table Api &amp&semi; SQL 翻译目录
Flink 官网 Table Api & SQL 相关文档的翻译终于完成,这里整理一个安装官网目录顺序一样的目录 [翻译]Flink Table Api & SQL —— Overv ...
【翻译】Flink Table Api &amp&semi; SQL —— 概念与通用API
本文翻译自官网:https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/common.html Flink Tabl ...

随机推荐

Spring的通知（Advice）
Spring提供了5种Advice类型: Interception Around:JointPoint前后调用 Before:JointPoint前调用 After Returning:JointPo ...
Python~删除空格，插入换行符号
f.write(rf.replace(' ','')) f.write(rf.replace('1041','\n1041')) 不能连续起作用? # -*- coding: UTF-8 -*- im ...
SQL学习笔记----更改SQL默认的端口号
1.SQLServer配置管理器----SQLServer网络配置----MSSQLSERVER的协议---TCP/IP(已启用)---IP地址清空素有的IP,在IPALL下更改默认的端口: 2. ...
转 scrollLeft,scrollWidth,clientWidth,offsetWidth之完全详解
scrollHeight: 获取对象的滚动高度. scrollLeft:设置或获取位于对象左边界和窗口中目前可见内容的最左端之间的距离 scrollTop:设置或获取位于对象最顶端和窗口中可见内容的最 ...
FTP服务器中vsftpd主配置文件解析
/etc/vsftpd/vsftpd.conf#################匿名权限控制############### anonymous_enable=YES #是否启用匿名用户no_anon_ ...
Python 中的重点来了：迭代器生成器
可迭代的对象:带有__iter__方法的,就是迭代器.字符串,列表,元祖,字典,集合,文件,都是可迭代的可迭代的对象使用了__iter__方法的,就是迭代器,带有__next__,__iter__方 ...
修改 Pattern代码使 Java 正则表达式支持下划线 &&num;39&semi;&lowbar;&&num;39&semi;
为什么由于工作是做数据ETL的,很多时候会使用到正则对数据进行提取,但是java的正则中的groupname不支持'_',官方的文档中是这样的: Group name A capturing gro ...
DotNetty网络通信框架学习
p{ text-align:center; } blockquote > p > span{ text-align:center; font-size: 18px; color: #ff0 ...
实时分析（在线查询）,firehose---clickhouse
firehose---clickhouse 在Hive中适不适合像传统数据仓库一样利用维度建模hive新功能 Cube, Rollup介绍https://blog.csdn.net/moon_yang ...
web状态管理机制
引入:b/s(浏览器/服务器模式)区别于winform的是winform中只加载一次页面构造函数,而b/s中只要点击按钮或者其他涉及后台的操作都会调用后台代码.一般情况下为了防止服务器过载,b/s不会 ...

正文