写点什么

【FlinkSQL】Flink SQL Query(三)- Join

用户头像
Alex🐒
关注
发布于: 2021 年 06 月 11 日

操作符

Join

Flink SQL 支持对动态表进行 Join 操作。有几种不同类型的 Join 可以支持多种语义。


默认情况下,Join 的顺序不会优化。表按 FROM 子句中指定的顺序联接。用户可以调整 Join 查询的性能,通过首先列出更新频率最低的表,最后列出更新频率最高的表。

常规 Join

常规 Join 是最通用的联接类型,其中任何新记录或对联接任一侧的更改都是可见的,并影响整个联接结果。例如,如果左侧有一个新记录,它将与右侧所有以前和将来的记录 Join。

SELECT * FROM OrdersINNER JOIN ProductON Orders.productId = Product.id
复制代码


对于流式查询,常规 Join 的语法是最灵活的,并且允许输入表任何类型的更新(插入、更新、删除)。但是,此操作具有重要的操作含义:要求将连接输入的两边永远保持在 Flink 状态(State)中。因此,计算查询结果所需的状态可能会无限增长,这取决于所有输入表和中间联接结果的不同输入行的数量。可以为查询配置提供适当的状态生存时间(TTL),以防止状态大小过大。请注意,这可能会影响查询结果的正确性。


对于流式查询,计算查询结果所需的状态可能会无限增长,具体取决于聚合类型和不同分组键的数量。请提供具有有效保留时间间隔的查询配置,以防止状态大小过大。

INNER JOIN

内连接语义,目前仅支持 equi-join,即 join 的联合条件至少拥有一个相等谓词。不支持任何 cross join 和 theta join。

SELECT *FROM OrdersINNER JOIN ProductON Orders.product_id = Product.id
复制代码

OUTER JOIN

外连接语义,Flink 支持 LEFT OUTER JOIN、RIGHT OUTER JOIN 和 FULL OUTER JOIN。目前仅支持 equi-join。

SELECT *FROM OrdersLEFT JOIN ProductON Orders.product_id = Product.id
SELECT *FROM OrdersRIGHT JOIN ProductON Orders.product_id = Product.id
SELECT *FROM OrdersFULL OUTER JOIN ProductON Orders.product_id = Product.id
复制代码

Interval Joins(时间区间链接)

Interval join 是常规 join 的子集,目前仅支持 equi-join,可以使用流的方式进行处理。


Interval join 需要至少一个相等谓词和一个限制了双方时间的 join 条件,比如,以下谓词是合法的 interval join 条件:

  • ltime = rtime

  • ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE

  • ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND


下面的示例,所有在收到后四小时内发货的 order 会与他们相关的 shipment 进行 join。

SELECT *FROM Orders o, Shipments sWHERE o.id = s.order_idAND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time
复制代码


对于流式查询,与常规 Join 相比,Interval join 只支持带有时间属性的 Append-only 表。由于时间属性是准单调递增的,Flink 可以在不影响结果正确性的情况下从其状态中删除旧值。

Temporal Joins

Temporal Joins 允许和版本表(Version/Temporal 表)进行 Join。这意味着可以在某个时间点检索值。左侧表可以是任意表,每一行与版本表(右侧表)对应行的版本进行 Join,语法如下:

SELECT [column_list]FROM table1 [AS <alias1>][LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ proctime | rowtime } [AS <alias2>]ON table1.column-name1 = table2.column-name1
复制代码

Event Time Temporal Join

例如,假设我们有一个订单表,每个订单都有不同货币的价格。为了正确地将此表规范化为单一货币(如美元),每个订单都需要从下订单时起使用正确的货币换算率。

-- Create a table of orders. This is a standard-- append-only dynamic table.CREATE TABLE orders (    order_id    STRING,    price       DECIMAL(32,2),    currency    STRING,    order_time  TIMESTAMP(3),    WATERMARK FOR order_time AS order_time) WITH (/* ... */);
-- Define a versioned table of currency rates. -- This could be from a change-data-capture-- such as Debezium, a compacted Kafka topic, or any other-- way of defining a versioned table. CREATE TABLE currency_rates ( currency STRING, conversion_rate DECIMAL(32, 2), update_time TIMESTAMP(3) METADATA FROM `values.source.timestamp` VIRTUAL WATERMARK FOR update_time AS update_time) WITH ( 'connector' = 'upsert-kafka', /* ... */);
SELECT order_id, price, currency, conversion_rate, order_time,FROM ordersLEFT JOIN currency_rates FOR SYSTEM TIME AS OF orders.order_timeON orders.currency = currency_rates.currency
order_id price currency conversion_rate order_time====== ==== ====== ============ ========o_001 11.11 EUR 1.14 12:00:00o_002 12.51 EUR 1.10 12:0600
复制代码

注意:事件时间 Temporal Join 由左右两侧的水印触发;请确保连接的两侧已正确设置水印。

注意:事件时间 Temporal Join 需要 Join 条件的等价条件中包含主键。

Processing Time Temporal Join

使用 processing time 属性,Join 将始终返回给定键的最新值。


下面的处理时间 Temporal Join 示例显示了表 LatestRates Join Append-only 表 orders。LatestRates 是维度表(例如 HBase 表)。在时间 10:15、10:30、10:52,LatestRates 表的内容如下:

10:15> SELECT * FROM LatestRates;
currency rate======== ======US Dollar 102Euro 114Yen 1
10:30> SELECT * FROM LatestRates;
currency rate======== ======US Dollar 102Euro 114Yen 1
10:52> SELECT * FROM LatestRates;
currency rate======== ======US Dollar 102Euro 116 <==== changed from 114 to 116Yen 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
复制代码


根据这些表,计算所有订单,转换为相同货币。

amount currency     rate   amount*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.amount, o.currency, r.rate, o.amount * r.rateFROM  Orders AS o  JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r  ON r.currency = o.currency
复制代码

查询使用处理时间时,左侧的每条记录都将与右侧表的当前版本相关联,新的 Order 记录总是与最新版本的 LatestRates 连接在一起。


查询结果对于处理时间来说是不确定的。处理时间 Temporal join 最常用于通过连接外部表(维度表)补充信息。

Lookup Join

Lookup Join 通常用于通过连接外部表(维度表)补充信息,要求一个表具有处理时间属性,另一个表使 Lookup Source Connector(查看 Connector 特性)。


Lookup join 使用上面的处理时间 Temporal Join 语法,右侧的表支持 Lookup Source Connector。

-- Customers is backed by the JDBC connector and can be used for lookup joinsCREATE TEMPORARY TABLE Customers (  id INT,  name STRING,  country STRING,  zip STRING) WITH (  'connector' = 'jdbc',  'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',  'table-name' = 'customers');
-- enrich each order with customer informationSELECT o.order_id, o.total, c.country, c.zipFROM Orders AS o JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c ON o.customer_id = c.id;
复制代码

在上面的示例中,Orders 表使用 MySQL 的 Customers 表的数据连接维度信息。在 Customer 的行更新时,联接结果不会更新。Lookup join 连接还需要一个强制的相等连接谓词(o.customer_id = c.id)。

Array Expansion(数组展开)

为给定数组中的每个元素返回新行

SELECT order_id, tagFROM Orders CROSS JOIN UNNEST(tags) AS t (tag)
复制代码

Table Function

连接表和 Table Function 的结果,左(外)表中的每一行将会与调用 Table Function 所产生的所有结果中相关行进行 join。用户自定义表函数( User-defined table functions,UDTFs ) 在执行前必须先注册。

INNER JOIN

若表函数返回了空结果,左表(outer)的行将会被删除。

SELECT order_id, resFROM Orders,LATERAL TABLE(table_func(order_id)) t(res)
复制代码

LEFT OUTER JOIN

若表函数返回了空结果,将会保留相对应的外部行并用空值填充结果。

SELECT order_id, resFROM OrdersLEFT OUTER JOIN LATERAL TABLE(table_func(order_id)) t(res)  ON TRUE
复制代码


发布于: 2021 年 06 月 11 日阅读数: 14
用户头像

Alex🐒

关注

还未添加个人签名 2020.04.30 加入

还未添加个人简介

评论

发布
暂无评论
【FlinkSQL】Flink SQL Query(三)- Join