Flink SQL联结与集合操作详解
一、前言
在上一篇文章中，我们学习了Flink SQL的聚合查询、窗口TVF和TopN等核心功能。本文将继续深入Flink SQL的查询能力，探讨联结（Join）操作、集合操作以及查询优化等高级特性。
联结查询是SQL中最常用也最复杂的操作之一，在流处理场景中更是如此。Flink SQL针对流处理的特点，提供了多种联结方式，包括常规联结（Regular Join）、间隔联结（Interval Join）和维表联结（Lookup Join）等。理解这些联结方式的原理和适用场景，是构建复杂流处理应用的关键。
二、联结（Join）查询
2.1 Flink SQL中的Join类型概览
Flink SQL中的联结查询大体上可以分为两类：SQL原生的联结查询方式，和流处理中特有的联结查询。
上图对比了Flink SQL中四种Join类型的特点：
| Join类型 | 联结条件 | 多版本构建 | 消费更新流 | 重要考虑因素 |
|---|---|---|---|---|
| Regular Join | 无限制 | 否 | 是 | 性能受probe side状态影响 |
| Lookup Join | 等值条件 | 否 | 是 | 可能非常慢，依赖外部系统可用性 |
| Temporal Join | 等值条件+完整主键 | 支持 | 是 | 事件时间必须正确定义 |
| Interval Join | 等值条件+时间范围 | 否 | 否 | 需要合理定义时间范围 |
2.2 常规联结（Regular Join）
常规联结（Regular Join）是SQL中原生定义的Join方式，是最通用的一类联结操作。它的具体语法与标准SQL的联结完全相同，通过关键字JOIN来联结两个表，后面用关键字ON来指明联结条件。
上图展示了Regular Join与Temporal Join在不同输入类型（append/upsert）下的输出行为差异。
Regular Join包含以下几种：
| Join类型 | 说明 | 输出行为 |
|---|---|---|
| Inner Join | 等值内联结 | 只有两条流Join到才输出 +[L, R] |
| Left Join | 左外联结 | 左流数据到达后，无论有没有Join到右流都输出 |
| Right Join | 右外联结 | 与Left Join逻辑相反 |
| Full Join | 全外联结 | 左流或右流数据到达后，无论有没有Join到都输出 |
等值内联结（INNER Equi-JOIN）：
SELECT*FROMwsINNERJOINws1ONws.id=ws1.id;目前仅支持等值联结条件。内联结会返回两表中符合联结条件的所有行的组合，也就是所谓的笛卡尔积（Cartesian product）中满足条件的部分。
等值外联结（OUTER Equi-JOIN）：
-- 左外联结SELECT*FROMwsLEFTJOINws1ONws.id=ws1.id;-- 右外联结SELECT*FROMwsRIGHTJOINws1ONws.id=ws1.id;-- 全外联结SELECT*FROMwsFULLOUTERJOINws1ONws.id=ws.id;Regular Join的注意事项：
- 实时Regular Join可以不是等值Join。等值Join和非等值Join区别在于，等值Join数据shuffle策略是Hash，会按照Join on中的等值条件作为id发往对应的下游；非等值Join数据shuffle策略是Global，所有数据发往一个并发
- 流的上游是无限的数据，Flink会将两条流的所有数据都存储在State中，所以Flink任务的State会无限增大，因此需要为State配置合适的TTL，以防止State过大
2.3 间隔联结（Interval Join）
间隔联结（Interval Join）返回符合约束条件的两条流中数据的笛卡尔积。与常规联结不同，间隔联结多了一个时间间隔的限制。
上图展示了Interval Join的时间窗口机制：对于Orders流中的每个事件，在Shipments流中查找落在时间窗口内的匹配事件。
语法要点：
- 两表的联结不需要用JOIN关键字，直接在FROM后将要联结的两表列出来，用逗号分隔
- 联结条件用WHERE子句来定义，用一个等值表达式描述
- 时间间隔限制在WHERE子句中通过AND追加
时间间隔定义方式：
-- 方式一：精确匹配ltime=rtime-- 方式二：范围匹配ltime>=rtimeANDltime<rtime+INTERVAL'10'MINUTE-- 方式三：BETWEEN语法ltimeBETWEENrtime-INTERVAL'10'SECONDANDrtime+INTERVAL'5'SECOND案例：
SELECT*FROMws,ws1WHEREws.id=ws1.idANDws.etBETWEENws1.et-INTERVAL'2'SECONDANDws1.et+INTERVAL'2'SECOND;2.4 维表联结（Lookup Join）
Lookup Join其实就是维表Join，实时获取外部缓存的Join。Lookup的意思就是实时查找。
上图展示了Lookup Join的工作原理：
- 左流（Source）的数据到达时，去外部维表（Paimon Table/MySQL等）中查找匹配数据
- 通过LRU缓存加速查询，减少对外部系统的访问压力
- 将关联后的结果输出到Sink
语法：
表AJOIN维度表名FORSYSTEM_TIMEASOF表A.proc_timeAS别名ONxx.字段=别名.字段案例：MySQL维表Join
-- 创建维表CREATETABLECustomers(idINT,name STRING,country STRING,zip STRING)WITH('connector'='jdbc','url'='jdbc:mysql://hadoop102:3306/customerdb','table-name'='customers');-- 使用Lookup JoinSELECTo.order_id,o.total,c.country,c.zipFROMOrdersASoJOINCustomersFORSYSTEM_TIMEASOFo.proc_timeAScONo.customer_id=c.id;Lookup Join的特点：
- 仅支持处理时间字段
- 左流（事实表）每来一条数据，都会去外部维表查找维度数据
- 如果维表数据发生变化，已经关联过的数据不会自动更新（基于处理时间快照）
三、Order By 和 Limit
3.1 Order By
Flink SQL支持Order By，但在实时任务中一般用的非常少。
实时任务中，Order By子句中必须要有时间属性字段，并且必须写在最前面且为升序。
SELECT*FROMwsORDERBYet,idDESC;3.2 Limit
SELECT*FROMwsLIMIT3;四、SQL Hints
4.1 什么是SQL Hints
在执行查询时，可以在表名后面添加SQL Hints来临时修改表属性，对当前job生效。
上图展示了SQL Hints在查询优化中的作用：通过Hints可以影响查询执行计划，优化查询性能。
4.2 Hints语法
select*fromws1/*+ OPTIONS('rows-per-second'='10')*/;常用Hints：
| Hint | 说明 |
|---|---|
OPTIONS('rows-per-second'='10') | 修改DataGen的生成速率 |
LOOKUP('table'='my_table2', 'async'='true') | 启用异步Lookup Join |
异步Lookup Join案例：
-- 同步Lookup（默认）SELECT/*+ LOOKUP('table'='my_table2', 'async'='false') */*FROMmy_table1ASt1JOINmy_table2FORSYSTEM_TIMEASOFt1.proctimeASt2ONt1.a=t2.c;-- 异步Lookup（提升吞吐量）SELECT/*+ LOOKUP('table'='my_table2', 'async'='true') */*FROMmy_table1ASt1JOINmy_table2FORSYSTEM_TIMEASOFt1.proctimeASt2ONt1.a=t2.c;五、集合操作
5.1 集合操作概述
Flink SQL支持标准SQL中的集合操作，包括UNION、UNION ALL、INTERSECT、INTERSECT ALL、EXCEPT和EXCEPT ALL。
上图通过维恩图直观展示了四种集合操作的结果集关系：
- UNION：合并两个集合并去重
- UNION ALL：合并两个集合不去重
- INTERSECT：取两个集合的交集
- EXCEPT：取左集合中不在右集合中的元素
5.2 UNION 和 UNION ALL
-- UNION：合并并去重(SELECTidFROMws)UNION(SELECTidFROMws1);-- UNION ALL：合并不去重(SELECTidFROMws)UNIONALL(SELECTidFROMws1);5.3 INTERSECT 和 INTERSECT ALL
-- INTERSECT：交集并去重(SELECTidFROMws)INTERSECT(SELECTidFROMws1);-- INTERSECT ALL：交集不去重(SELECTidFROMws)INTERSECTALL(SELECTidFROMws1);5.4 EXCEPT 和 EXCEPT ALL
-- EXCEPT：差集并去重(SELECTidFROMws)EXCEPT(SELECTidFROMws1);-- EXCEPT ALL：差集不去重(SELECTidFROMws)EXCEPTALL(SELECTidFROMws1);5.5 流式集合操作的特点
上述SQL在流式任务中，如果一条左流数据先来了，没有从右流集合数据中找到对应的数据时会直接输出，当右流对应数据后续来了之后，会下发回撤流将之前的数据给撤回。这也是一个回撤流。
5.6 IN 子查询
In子查询的结果集只能有一列：
SELECTid,vcFROMwsWHEREidIN(SELECTidFROMws1);上述SQL的In子句和之前介绍到的Inner Join类似。并且In子查询也会涉及到大状态问题，要注意设置State的TTL。
六、总结
本文详细讲解了Flink SQL中的联结与集合操作：
常规联结（Regular Join）：包括Inner/Left/Right/Full Join，语法与标准SQL一致，但需要注意流式场景下状态无限增长的问题
间隔联结（Interval Join）：在等值联结的基础上增加时间间隔限制，适合有时间范围关联需求的场景
维表联结（Lookup Join）：流与外部存储（MySQL/Redis/HBase等）的实时关联，仅支持处理时间
Order By/Limit：实时任务中Order By必须包含时间属性字段且放在最前面
SQL Hints：临时修改表属性，常用于优化Lookup Join（同步/异步）和DataGen参数
集合操作：UNION/UNION ALL、INTERSECT/INTERSECT ALL、EXCEPT/EXCEPT ALL，流式场景下会产生回撤流
In子查询：结果集只能有一列，底层类似于Inner Join，需要注意大状态问题
理解这些联结和集合操作的原理与适用场景，是构建复杂流处理SQL应用的基础。下一篇文章我们将继续深入Flink SQL的Connector与Catalog实战。
如果本文对你有帮助，欢迎点赞 👍 + 收藏 ⭐ + 关注 🔖，你的支持是我持续创作的动力！