在前一篇 《Flink SQL 窗口(Window)操作详解》 中我们已经打好了时间与窗口的基础。但在真实业务里单条流上的聚合往往只是第一步更常见的需求是把多条业务流关联起来一起看例如订单流 支付流衡量下单到付款的转化效果浏览流 下单流分析从曝光、点击到下单的完整漏斗用户行为流 用户画像维表驱动推荐、风控等在线决策这些需求背后的共性能力就是双流 JOIN。本文以「订单流 支付流」为主线从环境准备、建表、造数到 JOIN 查询一步步带你搞懂 Flink SQL 中的双流 JOIN 思路与实践。一、双流 JOIN 适用的典型场景订单与支付关联找出已下单但未支付、支付失败等情况广告曝光与点击关联计算点击率、转化路径日志与告警规则关联实时检测异常行为这些场景有两个共同特征两条都是事实流不断追加的新事件需要在时间范围内去匹配事件谁先发生、允许多长时间内匹配因此在流计算中做 JOIN一定绕不开时间字段和水位线Watermark。二、Flink 中常见的 JOIN 类型在 Flink SQL 的流模式下常见的双流关联方式有普通 JOIN基于等值条件 时间字段的 JOINInterval Join基于「时间区间」的双流 JOINTemporal Join一条流 维表变更流的时态关联本篇主要聚焦前两种更贴近「订单流 支付流」这样的事实双流场景。三、准备示例数据表安装 Kafka环境前提在 WSL2 的 Ubuntu 环境中安装并启动 Kafka请参考 《从零开始学Flink数据源》。安装 Flink Kafka SQL Connector需要把 Flink 的 Kafka SQL Connector JAR 包放到$FLINK_HOME/lib目录下。以本系列示例使用的Flink 1.20.1 Kafka 3.4.0-1.20为例可以这样操作确认你的 Flink 安装目录假设为/opt/flink/* by 01130.hk - online tools website : 01130.hk/zh/allencrypt.html */ export FLINK_HOME/opt/flink下载 Kafka SQL Connector JAR 到 Flink 的lib目录/* by 01130.hk - online tools website : 01130.hk/zh/allencrypt.html */ cd $FLINK_HOME/lib wget https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.4.0-1.20/flink-sql-connector-kafka-3.4.0-1.20.jar如果你是 Windows WSL2可以在 WSL2 里执行同样的命令或者用浏览器下载后手动拷贝到lib目录。如果你使用的是独立集群或远程集群需要重启 Flink 集群让新 JAR 在 JobManager/TaskManager 上生效cd $FLINK_HOME bin/stop-cluster.sh bin/start-cluster.sh如果只是本地直接运行bin/sql-client.sh启动内嵌 mini-cluster则只需重启 SQL Client 即可。启动 Flink SQL Client然后执行本文后续的建表与查询示例cd $FLINK_HOME bin/sql-client.sh准备 Kafka 中的示例数据表我们假设已经从 Kafka 中读取两条流orders订单流payments支付流并在建表时定义了事件时间和水位线CREATE TABLE orders ( order_id STRING, user_id STRING, order_amount DECIMAL(10, 2), order_time TIMESTAMP_LTZ(3), WATERMARK FOR order_time AS order_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic orders, properties.bootstrap.servers 127.0.0.1:9092, properties.group.id flink-orders, scan.startup.mode earliest-offset, format json, json.timestamp-format.standard ISO-8601 ); CREATE TABLE payments ( pay_id STRING, order_id STRING, pay_amount DECIMAL(10, 2), pay_time TIMESTAMP_LTZ(3), WATERMARK FOR pay_time AS pay_time - INTERVAL 5 SECOND ) WITH ( connector kafka, topic payments, properties.bootstrap.servers 127.0.0.1:9092, properties.group.id flink-payments, scan.startup.mode earliest-offset, format json, json.timestamp-format.standard ISO-8601 );有了时间字段和水位线Flink 才能在流模式下安全地做双流 JOIN并在「时间窗」关闭后清理状态。使用 Kafka Console Producer 造测试数据上面的 DDL 建好了orders和payments两张表对应的是 Kafka 中的两个 Topic。接下来我们用 Kafka 自带的命令行工具写入几条 JSON 测试数据。假设你已经在 WSL2 的 Ubuntu 中启动好了 Kafka包括 ZooKeeper 或 KRaft进入 Kafka 安装目录执行1. 往订单 Topic 写入数据bin/kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic orders在命令行中输入几条 JSON 数据按回车发送一条{order_id:o_1,user_id:u_1,order_amount:100.00,order_time:2026-02-16T14:41:00Z} {order_id:o_2,user_id:u_2,order_amount:200.00,order_time:2026-02-16T14:42:00Z} {order_id:o_3,user_id:u_1,order_amount:150.00,order_time:2026-02-16T14:45:00Z}2. 往支付 Topic 写入数据新开一个终端同样进入 Kafka 安装目录执行bin/kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic payments输入对应的支付 JSON 数据{pay_id:p_1,order_id:o_1,pay_amount:100.00,pay_time:2026-02-16T14:41:00Z} {pay_id:p_2,order_id:o_2,pay_amount:200.00,pay_time:2026-02-16T14:42:00Z}这里的字段名、时间格式都要和前面建表时定义的一致这样 Flink 才能正确反序列化 JSON 并进行双流 JOIN。四、基于时间条件的普通双流 JOIN先来看最直观的一种写法同时指定「关联键」和「时间范围」。需求统计订单在下单后 15 分钟内完成支付的记录。SELECT o.order_id, o.user_id, o.order_amount, o.order_time, p.pay_id, p.pay_amount, p.pay_time FROM orders AS o JOIN payments AS p ON o.order_id p.order_id AND p.pay_time BETWEEN o.order_time AND o.order_time INTERVAL 15 MINUTE;这里有几点非常关键o.order_id p.order_id以订单号作为两条流的业务主键pay_time BETWEEN order_time AND order_time INTERVAL 15 MINUTE明确限定“下单后 15 分钟内支付”这类时间约束使用事件时间字段配合水位线可以在保证计算正确性的前提下控制状态大小并处理一定范围内的迟到数据如果你希望保留那些下单了但超时未支付的记录可以将上面的JOIN改为LEFT JOIN然后在下游以p.pay_id IS NULL作为“未支付/超时”的判断条件。五、Interval Join显式时间区间的双流 JOIN普通 JOIN 中的时间条件本质上就是一种「区间约束」。在 Flink Table API 中有一个更明确的概念Interval Join。等价的 Interval Join 写法大致如下Table API 伪代码仅作为概念理解SELECT o.order_id, o.order_time, p.pay_id, p.pay_time FROM orders AS o JOIN payments AS p ON o.order_id p.order_id AND p.pay_time BETWEEN o.order_time AND o.order_time INTERVAL 15 MINUTE;无论是普通 JOIN 还是 Interval Join本质上都是以某个时间字段作为「对齐基准」设定一个前后允许的时间区间在这个区间内匹配到的记录会输出为 JOIN 结果六、迟到数据与状态清理在流式 JOIN 中最容易被忽略但又非常重要的一点就是状态会不断累积。Flink 会根据时间条件和水位线来决定某条历史事件是否还有可能再匹配到另一条流的事件超出时间范围且水位线已推进时可以安全地清理对应状态设计双流 JOIN 时建议考虑时间窗口不要设置得过大否则状态会膨胀根据业务的真实延迟来设置水位线与时间区间对于极端迟到的数据是丢弃、旁路输出还是通过补偿机制处理七、一个完整的小结通过本文你需要记住下面几点双流 JOIN 场景非常常见本质是两条事实流在时间上的匹配流式 JOIN 一定要依赖事件时间 水位线来控制状态和迟到数据常见的方式包括基于时间条件的普通 JOIN 和 Interval JOIN设计时间区间时要在「业务容忍度」和「资源消耗」之间做权衡在下一篇中我们可以继续围绕「实时数仓」或「维表时态 Join」展开把事实流与维度数据关联起来构建更真实的 Flink SQL 实战项目。原文来自:http://blog.daimajiangxin.com.cn