Flink join详解

Flink SQL支持对动态表进行复杂而灵活的连接操作。 为了处理不同的场景，需要多种查询语义，因此有几种不同类型的 Join。

默认情况下，joins 的顺序是没有优化的。表的 join 顺序是在FROM从句指定的。可以通过把更新频率最低的表放在第一个、频率最高的放在最后这种方式来微调 join 查询的性能。需要确保表的顺序不会产生笛卡尔积，因为不支持这样的操作并且会导致查询失败。

1. Regular Joins

Regular join 是最通用的 join 类型。在这种 join 下，join 两侧表的任何新记录或变更都是可见的，并会影响整个 join 的结果。 例如：如果左边有一条新纪录，在Product.id相等的情况下，它将和右边表的之前和之后的所有记录进行 join。

• window join 和 interval Join 都是基于划分窗口，将无界数据变为有界数据进行关联
• regular join 则还是基于无界数据进行关联

SELECT * FROM OrdersINNER JOIN ProductON Orders.productId = Product.id

对于流式查询，regular join 的语法是最灵活的，允许任何类型的更新（插入、更新、删除）输入表。 然而，这种操作具有重要的操作意义：Flink 需要将 Join 输入的两边数据永远保持在状态中。 因此，计算查询结果所需的状态可能会无限增长，这取决于所有输入表的输入数据量。你可以提供一个合适的状态 time-to-live (TTL) 配置来防止状态过大。注意：这样做可能会影响查询的正确性。查看查询配置了解详情。

以 A 流 left join B 流举例

A 流数据到来之后，直接去尝试关联 B 流数据：

• 如果关联到了则直接下发关联到的数据
• 如果没有关联到则也直接下发没有关联到的数据，后续 B 流中的数据到来之后，会把之前下发下去的没有关联到数据撤回，然后把关联到的数据数据进行下发。由此可以看出这是基于 Flink SQL 的 retract 机制，则也就说明了其目前只支持 Flink SQL

INNER Equi-JOIN

根据 join 限制条件返回一个简单的笛卡尔积。目前只支持 equi-joins，即：至少有一个等值条件。不支持任意的 cross join 和 theta join。(cross join 指的是类似SELECT * FROM table_a CROSS JOIN table_b，theta join 指的是类似SELECT * FROM table_a, table_b）

SELECT *FROM OrdersINNER JOIN ProductON Orders.product_id = Product.id

OUTER Equi-JOIN

返回所有符合条件的笛卡尔积（即：所有通过 join 条件连接的行），加上所有外表没有匹配到的行。Flink 支持 LEFT、RIGHT 和 FULL outer joins。目前只支持 equi-joins，即：至少有一个等值条件。不支持任意的 cross join 和 theta join。

SELECT *FROM OrdersLEFT JOIN ProductON Orders.product_id = Product.idSELECT *FROM OrdersRIGHT JOIN ProductON Orders.product_id = Product.idSELECT *FROM OrdersFULL OUTER JOIN ProductON Orders.product_id = Product.id

RegularJoin特点

SQL API

CREATE TABLE show_log_table (log_id BIGINT,show_params STRING) WITH ('connector' = 'datagen','rows-per-second' = '1','fields.show_params.length' = '3','fields.log_id.min' = '1','fields.log_id.max' = '10');CREATE TABLE click_log_table (log_id BIGINT,click_params STRING)WITH ('connector' = 'datagen','rows-per-second' = '1','fields.click_params.length' = '3','fields.log_id.min' = '1','fields.log_id.max' = '10');CREATE TABLE sink_table (s_id BIGINT,s_params STRING,c_id BIGINT,c_params STRING) WITH ('connector' = 'print');INSERT INTO sink_tableSELECTshow_log_table.log_id as s_id,show_log_table.show_params as s_params,click_log_table.log_id as c_id,click_log_table.click_params as c_paramsFROM show_log_tableLEFT JOIN click_log_table ON show_log_table.log_id = click_log_table.log_id;

DataStream API

import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class RegularJoinExample {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 创建两个数据流DataStream leftStream = ...; // 左侧数据流DataStream rightStream = ...; // 右侧数据流// Regular Join 操作DataStream joinedStream = leftStream.join(rightStream).where(leftRecord -> leftRecord.getKey()) // 指定左侧数据流的连接键.equalTo(rightRecord -> rightRecord.getKey()) // 指定右侧数据流的连接键.window(TumblingProcessingTimeWindows.of(Time.seconds(10))) // 指定窗口，如果不指定，默认为全局窗口.apply((leftRecord, rightRecord) -> new ResultRecord(leftRecord, rightRecord)); // 指定连接后的处理逻辑// 打印结果joinedStream.print();// 执行任务env.execute("Regular Join Example");}}

RegularJoin特点

• join的流的数据都会存在flink的状态中Forever
• 流中所有的数据对对方来说都是可见的
• 只能用于有界流
• 只能用于等值连接
• 支持 Flink SQL

解决方案的适用场景
该种解决方案虽然是目前在产出质量、时效性上最好的一种解决方案，但是在实际场景中使用时，也存在一些问题：

⭐ 基于 retract 机制，所有的数据都会存储在 state 中以判断能否关联到，所以我们要设置合理的 state ttl 来避免大 state 问题导致的任务不稳定
⭐ 基于 retract 机制，所以在数据发生更新时，会下发回撤数据、最新数据 2 条消息，当我们的关联层级越多，则下发消息量的也会放大
⭐ sink 组件要支持 retract，我们不要忘了最终数据是要提供数据服务给需求方进行使用的，所以我们最终写入的数据组件也需要支持 retract，比如 MySQL。如果写入的是 Kafka，则下游消费这个 Kafka 的引擎也需要支持回撤\更新机制。

Join优化思路

针对上面 3 节说到的 Flink Join 的方案，各自都有一些优势和劣势存在。

但是我们可以发现，无论是哪一种 Join 方案，Join 的前提都是将 A 流和 B 流的数据先存储在状态中，然后再进行关联。

即在实际生产中使用时常常会碰到的问题就是：大状态的问题。

两种大Key状态优化方案

关于大状态问题业界常见两种解决思路：

⭐ 减少状态大小：在 Flink Join 中的可以想到的优化措施就是减少 state key 的数量。在未优化之前 A 流和 B 流的数据往往是存储在单独的两个 State 实例中的，那么我们的优化思路就是将同 Key 的数据放在一起进行存储，一个 key 的数据只需要存储一份，减少了 key 的数量

⭐ 转移状态至外存：大 State 会导致 Flink 任务不稳定，那么我们就将 State 存储在外存中，让 Flink 任务轻量化，比如将数据存储在 Redis 中，A 流和 B 流中相同 key 的数据共同维护在一个 Redis 的 hashmap 中，以供相互进行关联

接下来看看这两种方案实际需要怎样落地。讲述思路也是按照以下几点进行阐述：
⭐ 优化方案说明
⭐ 优化方案 Flink API
⭐ 优化方案的特点
⭐ 优化方案的适用场景

Flink Join 优化方案：同 key 共享 State
优化方案说明
将两条流的数据使用 union、connect 算子合并在一起，然后使用一个共享的 state 进行处理

优化方案 Flink API

FlinkEnv flinkEnv = FlinkEnvUtils.getStreamTableEnv(args);flinkEnv.env().setParallelism(1);flinkEnv.env().addSource(new SourceFunction