FlinkSql 的 Join
Flink 官网将其分为了 Joins 和 Window Joins两个大类，其中里面又分了很多 Join 方式
参考文档：
Joins | Apache Flink
Window JOIN | Apache Flink
Joins
官网介绍共有6种方式：
-
Regular Join：流与流的 Join，包括 Inner Join、Outer Equal Join
-
Interval Join：流与流的 Join，两条流一段时间区间内的 Join
-
Temporal Join：流与流的 Join，包括事件时间，处理时间的 Temporal Join，类似于离线中的快照 Join
-
Lookup Join：流与外部维表的 Join
-
Array Expansion：表字段的列转行，类似于 Hive 的 explode 数据炸开的列转行
-
Table Function：自定义函数的表字段的列转行，支持 Inner Join 和 Left Outer Join
Regular Join
写法上和传统数据库没有区别，关联条件支持等值和非等值Join，有Inner Join 和 Outer Join（Left Join、Right Join、FULL JOIN）
有人问我为什么要特别区分内外连接，后面会用到
内连接是通过匹配两个表之间的共同列，返回满足连接条件的行。只有在连接条件匹配的情况下，才会返回结果。
外连接是在内连接的基础上，还包括了不满足连接条件的行。
SELECT order_id, uid, price, user_name
FROM order a
Left JOIN user b
ON a.uid = b.uid
顺便了解一下流是怎么 Join 的：
和离线不同，离线是一批数据一起运算的，完成后输出结果
FlinkSql是Dynamic Table的概念，数据在 State 里面，每来一条数据就会对左右两边的数据进行关联
Regular Join 的 State 默认是永久保存的，为了避免 State 无限膨胀，可以根据情况决定是否设置状态清理：table.exec.state.ttl（目前是根据更新时间来判断是否过期，而非访问时间）
再来看看几种 Join ，其中outer Join产生的回撤流是和传统离线方式有很大区别的：
首先不考虑数据源有回撤的情况，Regular Join在 Outer Join 时会产生回撤流，L-左表、R-右表
-
Inner Join：两条流 Join 到才输出
+[L, R]，关联不上不会输出 -
Left Join：当左流数据到达之后就会直接输出
可以 Join 到右流则输出 +[L,R]，Join 不到右流输出 +[L,null]）
如果之后右流之后数据到达之后，发现左流之前输出过没有 Join 到的数据
则会发起回撤流，先输出 -[L,null]，然后在输出一条 +[L,R]
-
Right Join：有 Left Join 一样，只是逻辑相反
-
Full Join：和Left原理一样，左流或者右流的数据到达之后，无论有没有 Join 到另外一条流的数据，都会输出，如果一条流的数据到达之后，发现之前另一条流之前输出过没有 Join 到的数据，则会发起回撤流
对右流来说：Join 到输出 +[L,R]，没 Join 到输出 +[null,R]，左流数据到达后回撤 -[null,R]，输出 +[L,R]
对左流来说：Join 到输出 +[L,R]，没 Join 到输出 +[L,null]），右流数据到达后回撤 -[L, null]，输出 +[L,R]
图解：
inner join 和 lef join 输出结果示例：
inner join
+I[5, d, 5, f]
+I[5, d, 5, 8]
+I[3, 4, 3, 0]
left join
+I[3, 4ab, null, null]
+I[5, f3c, 5, c05]
+I[5, 6e2, 5, c05]
-D[3, 4ab, null, null]
+I[3, 4ab, 3, 765]关于 Regular Join 的注意事项：
-
实时 Regular Join 可以不是
等值 join。等值 join和非等值 join区别在于，等值 join数据 shuffle 策略是 Hash，会按照 Join on 中的等值条件作为 id 发往对应的下游；非等值 join数据 shuffle 策略是 Global，所有数据发往一个并发，按照非等值条件进行关联 -
Join 的流程是左流新来一条数据之后，会和右流中符合条件的所有数据做 Join，然后输出，如果是outer join会立即输出之后产生回撤流
-
流的上游是无限的数据，所以要做到关联的话，Flink 会将两条流的所有数据都存储在 State 中，所以 Flink 任务的 State 会无限增大，因此你需要为 State 配置合适的 TTL，以防止 State 过大。
Interval Join
Interval Join 只支持普通 Append 数据流，不支持含 Retract 的动态表
Interval Join 左右表仅在某个时间范围（给定上界和下界）内进行关联，这个时间区间支持event time 和 processing time两种语义，如果是 event time，会根据区间和Watermark自动清理状态
场景示例：用户下单产生订单信息，用户必须在下单后一个小时以内付款，输出付款的订单信息
SELECT
o.orderId,
o.productName,
p.payType,
o.orderTime，
cast(payTime as timestamp) as payTime
FROM Orders o
JOIN Payment p
ON o.orderId = p.orderId
AND p.payTime BETWEEN orderTime AND orderTime + INTERVAL ‘1’ HOURInterval Join 几种方式，需要注意 Interval Join 不会产生回撤流：
-
Inner Join：只有两条流 Join 到才输出，输出
+[L, R] -
Left Join：和 Regular Join 不同，左流数据到达之后，如果没有 Join 到右流的数据，就会等待（放在 State 中等），如果之后右流之后数据到达之后，发现能和刚刚那条左流数据 Join 到，这时输出
+[L, R]。事件时间中随着 Watermark 的推进（也支持处理时间）。如果发现发现左流 State 中的数据过期了，就把左流中过期的数据从 State 中删除，然后输出+[L, null]（这时候其实已经延迟了），如果右流 State 中的数据过期了，就直接从 State 中删除 -
Right Join：同 Left Join，逻辑相反
-
Full Join：流任务中，左流或者右流的数据到达之后，如果没有 Join 到另外一条流的数据，就会等待（左流放在左流对应的 State 中等，右流放在右流对应的 State 中等），如果之后另一条流数据到达之后，发现能和刚刚那条数据 Join 到，则会输出
+[L, R]。事件时间中随着 Watermark 的推进（也支持处理时间），发现 State 中的数据能够过期了，就将这些数据从 State 中删除并且输出（左流过期输出+[L, null]，右流过期输出-[null, R]）
图解：
inner join不用多说，看看 left join 输出结果示例：
+I[6, e, 6, 7]
+I[11, d, null, null]
+I[7, b, null, null]
+I[8, 0, 8, 3]
+I[13, 6, null, null]关于 Interval Join 的注意事项：
-
实时 Interval Join 可以不是 等值 join。等值 join 和 非等值 join 区别在于，等值 join 数据 shuffle 策略是 Hash，会按照 Join on 中的等值条件作为 id 发往对应的下游；非等值 join 数据 shuffle 策略是 Global，所有数据发往一个并发，然后将满足条件的数据进行关联输出
-
outer join 不会产生回撤流，关联不上会在 State 过期时发送数据，会有延迟
Temporal Joins
这种关联方式同样是传统数据库没有的，但是会发现和数仓的拉链表Join有点类似
Temporal Join 支持和 Verisoned Table 进行关联，也支持 event time 和 processing time 两种语义，支持inner join 和 left join 两种方式
事件时间 ，在解决多版本问题时有奇效：
-
事件时间的 Temporal Join 一定要给左右两张表都设置 Watermark
-
事件时间的 Temporal Join 一定要把 Versioned Table 的主键包含在 Join on 的条件中
--官网案例
CREATE TABLE orders (
order_id STRING,
price DECIMAL(32,2),
currency STRING,
order_time TIMESTAMP(3),
WATERMARK FOR order_time AS order_time - INTERVAL '15' SECOND
) WITH (/* ... */);
-- 必须定义一个 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 - INTERVAL '15' SECOND,
PRIMARY KEY(currency) NOT ENFORCED
) WITH (
'connector' = 'kafka'
/* ... */
);
SELECT
order_id,
price,
orders.currency,
conversion_rate,
order_time
FROM orders
LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time
ON orders.currency = currency_rates.currency;
order_id price currency conversion_rate order_time
======== ===== ======== =============== =========
o_001 11.11 EUR 1.14 12:00:00
o_002 12.51 EUR 1.10 12:06:00Flink SQL 会为 Versioned Table 维护 Primary Key 下的所有历史时间版本的数据，然后根据左表Orders的事件时间关联到对应时间的 Versioned Table 的汇率
Processing Time，由于是处理时间，只维护了最新的状态数据，不需要关心历史版本的数据，直接根据LeftTable数据到达的时间关联最新的数据
另外还支持 Temporal Table Functionv Join，但是一般不怎么用（至少我基本不这样写）
SELECT
o_amount, r_rate
FROM
Orders,
LATERAL TABLE (Rates(o_proctime))
WHERE
r_currency = o_currencyLookup Join
Lookup Join 通常用于关联外部系统数据（比如Mysql、Hbase等）,但是目前只支持 processing time，只能以处理时间关联最新的数据（这个最新是有代价的）
实际用起来其实会发现功能上和 version table 的processing 类似
-- 官网案例，需要定义一个外部存储的表
CREATE 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 information
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;待办：lookup支持cache，cache的异步查询原理，数据更新的延迟，参数调优等等
Array Expansion
常见的用法就是类似Spark 的 lateral view expload(arr)
SELECT order_id, tag
FROM Orders CROSS JOIN UNNEST(tagArray) AS t (tag)Table Function
其实和 Array Expansion 功能类似，但是 Table Function 本质上是个 UDTF 函数，并且支持自定义函数
Window Joins
见 FlinkSql 窗口函数
语法示例：
SELECT L.num as L_Num, L.id as L_Id, R.num as R_Num, R.id as R_Id,
COALESCE(L.window_start, R.window_start) as window_start,
COALESCE(L.window_end, R.window_end) as window_end
FROM (
SELECT * FROM TABLE(TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
) L
INNER JOIN (
SELECT * FROM TABLE(TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
) R
ON L.num = R.num AND L.window_start = R.window_start AND L.window_end = R.window_end;SELECT *
FROM (
SELECT * FROM TABLE(TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
) L WHERE EXISTS (
SELECT * FROM (
SELECT * FROM TABLE(TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
) R WHERE L.num = R.num AND L.window_start = R.window_start AND L.window_end = R.window_end);
