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类型的特点：

2.2 常规联结（Regular Join）

常规联结（Regular Join）是SQL中原生定义的Join方式，是最通用的一类联结操作。它的具体语法与标准SQL的联结完全相同，通过关键字JOIN来联结两个表，后面用关键字ON来指明联结条件。

上图展示了Regular Join与Temporal Join在不同输入类型（append/upsert）下的输出行为差异。

Regular 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：

异步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中的联结与集合操作：

1. 常规联结（Regular Join）：包括Inner/Left/Right/Full Join，语法与标准SQL一致，但需要注意流式场景下状态无限增长的问题

2. 间隔联结（Interval Join）：在等值联结的基础上增加时间间隔限制，适合有时间范围关联需求的场景

3. 维表联结（Lookup Join）：流与外部存储（MySQL/Redis/HBase等）的实时关联，仅支持处理时间

4. Order By/Limit：实时任务中Order By必须包含时间属性字段且放在最前面

5. SQL Hints：临时修改表属性，常用于优化Lookup Join（同步/异步）和DataGen参数

6. 集合操作：UNION/UNION ALL、INTERSECT/INTERSECT ALL、EXCEPT/EXCEPT ALL，流式场景下会产生回撤流

7. In子查询：结果集只能有一列，底层类似于Inner Join，需要注意大状态问题

理解这些联结和集合操作的原理与适用场景，是构建复杂流处理SQL应用的基础。下一篇文章我们将继续深入Flink SQL的Connector与Catalog实战。

如果本文对你有帮助，欢迎点赞 👍 + 收藏 ⭐ + 关注 🔖，你的支持是我持续创作的动力！