Flink Join 核心解析：类型、原理、实操
Flink 中的 Join 是用于关联两个或多个数据流 / 数据集的核心操作，分为批处理（Batch）Join和流处理（Stream）Join两大类。批处理 Join 基于静态数据集，逻辑与传统数据库 Join 类似；而流处理 Join 基于无限数据流，需结合窗口或状态实现，是 Flink 流处理的重点和难点。以下是详细解析。
一、Flink Join 的核心分类
Flink Join 主要分为两大维度：
| 维度 | 类型 | 适用场景 | 核心原理 |
|---|---|---|---|
| 数据形态 | 批处理 Join（DataSet API） | 静态数据集关联（如历史数据） | 基于内存 / 磁盘的哈希 Join、排序合并 Join |
| 数据形态 | 流处理 Join（DataStream API） | 实时数据流关联（如实时订单 + 用户） | 基于窗口 / 状态存储，在窗口内关联数据 |
| Join 逻辑 | Inner Join（内连接） | 只保留两边都匹配的数据 | 仅输出匹配的记录 |
| Join 逻辑 | Left Outer Join（左外连接） | 保留左表所有数据，右表匹配的数据 | 左表无匹配时，右表字段为 null |
| Join 逻辑 | Right Outer Join（右外连接） | 保留右表所有数据，左表匹配的数据 | 右表无匹配时，左表字段为 null |
| Join 逻辑 | Full Outer Join（全外连接） | 保留两边所有数据 | 无匹配时对应字段为 null |
注意：Flink 1.12+ 主推 Table API/SQL 实现 Join，其语法更简洁、功能更完善（支持更多 Join 类型），而 DataStream API 需手动实现窗口 / 状态管理。
二、批处理 Join（DataSet API）
批处理 Join 针对静态数据集，语法和逻辑与传统数据库 Join 一致，Flink 会根据数据量自动选择哈希 Join或排序合并 Join优化性能。
1. 核心原理
- 哈希 Join：将小数据集加载到内存中构建哈希表，遍历大数据集进行匹配（适用于小表 + 大表）；
- 排序合并 Join：将两个数据集排序后，遍历并合并匹配的数据（适用于大表 + 大表）。
2. 代码实操（Java）
import org.apache.flink.api.common.functions.JoinFunction; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple3; public class BatchJoinExample { public static void main(String[] args) throws Exception { // 1. 创建批处理执行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 2. 定义左表：用户数据（用户ID，用户名） DataSet<Tuple2<Integer, String>> userDataSet = env.fromElements( Tuple2.of(1, "Alice"), Tuple2.of(2, "Bob"), Tuple2.of(3, "Charlie") ); // 3. 定义右表：订单数据（订单ID，用户ID，金额） DataSet<Tuple3<Integer, Integer, Double>> orderDataSet = env.fromElements( Tuple3.of(101, 1, 99.9), Tuple3.of(102, 2, 199.9), Tuple3.of(103, 4, 299.9) // 用户ID=4，左表无匹配 ); // 4. Inner Join：按用户ID关联，保留两边匹配的数据 DataSet<Tuple3<Integer, String, Double>> innerJoinResult = userDataSet .join(orderDataSet) .where(0) // 左表的关联字段：用户ID（Tuple2的第0位） .equalTo(1) // 右表的关联字段：用户ID（Tuple3的第1位） .with(new JoinFunction<Tuple2<Integer, String>, Tuple3<Integer, Integer, Double>, Tuple3<Integer, String, Double>>() {