在FLink中经常会遇到双流关联的的场景比如订单流和支付流的关联用户行为流和用户信息流等。那么在Flink中都提供了哪些双流的关联呢下面主要从DataStream和Flink SQL角度入手说明。DataStreamWindow Join窗口Join两条流在同一个时间窗口内进行Join窗口结束时出触发计算。// 设置10秒的窗口期stream1.join(stream2).where(r-r.id).equalTo(r-r.id).window(TumblingEventTimeWindows.of(Time.seconds(10))).apply((e1,e2)-e1 e2);Window Join的特点只能做 inner join两条流的数据必须落在同一个窗口内才能匹配支持 Tumbling / Sliding / Session WindowWindow Join只能做 inner joinjoin不上的数据会直接丢弃没有机制输出到侧输出流所以谨慎使用。Interval Join区间Join基于事件时间允许一条流的数据在另一条流数据的时间范围内进行匹配。// stream1流的一个记录可以在stream2前后5秒的这个区间匹配stream1.keyBy(r-r.id).intervalJoin(stream2.keyBy(r-r.id)).between(Time.seconds(-5),Time.seconds(5)).process(newProcessJoinFunction(){OverridepublicvoidprocessElement(Leftl,Rightr,Contextctx,CollectorStringout){out.collect(l r);}});Interval Join的特点只支持 inner join只支持 event time比 Window Join 更灵活适合时间偏差场景同样Interval Join只能做 inner joinjoin不上的数据会直接丢弃没有机制输出到侧输出流所以谨慎使用。CoProcessFunction底层 APICoprocessFunction是最灵活的方式可以实现inner / left / right / full outer join以及各种自定义逻辑。stream1.keyBy(r-r.id).connect(stream2.keyBy(r-r.id)).process(newCoProcessFunctionLeft,Right,String(){// 用 State 缓存两侧数据自行实现匹配逻辑MapStateString,LeftleftState;MapStateString,RightrightState;OverridepublicvoidprocessElement1(Leftl,Contextctx,CollectorStringout){// 查 rightState匹配则输出否则存入 leftState}OverridepublicvoidprocessElement2(Rightr,Contextctx,CollectorStringout){// 查 leftState匹配则输出否则存入 rightState}});CoProcessFunction的特点完全自定义支持所有 join 类型需要手动管理 State 和 TTL防止状态无限增长适合复杂业务逻辑使用CoProcessFunction实现Left Join下面使用CoProcessFunction实现Left Join。publicclassLeftJoinExample{// 左流数据DataAllArgsConstructorpublicstaticclassOrder{publicStringorderId;publicStringuserId;publiclongtimestamp;}// 右流数据DataAllArgsConstructorpublicstaticclassPayment{publicStringorderId;publicdoubleamount;publiclongtimestamp;}// 输出结果DataAllArgsConstructorpublicstaticclassJoinResult{publicStringorderId;publicStringuserId;publicDoubleamount;// 右流未匹配时为 null}publicstaticclassLeftJoinFunctionextendsCoProcessFunctionOrder,Payment,JoinResult{// 未匹配的左流数据privateMapStateString,OrderorderState;// 右流数据缓存privateMapStateString,PaymentpaymentState;// 等待匹配的超时时间privatestaticfinallongJOIN_TIMEOUT10_000L;Overridepublicvoidopen(Configurationparameters){orderStategetRuntimeContext().getMapState(newMapStateDescriptor(order-state,String.class,Order.class));paymentStategetRuntimeContext().getMapState(newMapStateDescriptor(payment-state,String.class,Payment.class));}OverridepublicvoidprocessElement1(Orderorder,Contextctx,CollectorJoinResultout)throwsException{PaymentpaymentpaymentState.get(order.orderId);if(payment!null){// 右流已有数据直接 join 输出out.collect(newJoinResult(order.orderId,order.userId,payment.amount));paymentState.remove(order.orderId);}else{// 右流暂无数据缓存左流注册超时定时器orderState.put(order.orderId,order);ctx.timerService().registerEventTimeTimer(order.timestampJOIN_TIMEOUT);}}OverridepublicvoidprocessElement2(Paymentpayment,Contextctx,CollectorJoinResultout)throwsException{OrderorderorderState.get(payment.orderId);if(order!null){// 左流已有数据直接 join 输出out.collect(newJoinResult(order.orderId,order.userId,payment.amount));orderState.remove(payment.orderId);}else{// 左流暂无数据缓存右流等待paymentState.put(payment.orderId,payment);}}OverridepublicvoidonTimer(longtimestamp,OnTimerContextctx,CollectorJoinResultout)throwsException{// 超时触发左流数据仍未匹配输出 null 右侧left join 语义for(Map.EntryString,Orderentry:orderState.entries()){Orderorderentry.getValue();if(order.timestampJOIN_TIMEOUTtimestamp){out.collect(newJoinResult(order.orderId,order.userId,null));orderState.remove(entry.getKey());}}}}publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();DataStreamOrderorderStreamenv.fromElements(newOrder(o1,u1,1000L),newOrder(o2,u2,2000L)).assignTimestampsAndWatermarks(WatermarkStrategy.OrderforBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((e,t)-e.timestamp));DataStreamPaymentpaymentStreamenv.fromElements(newPayment(o1,99.9,3000L)// o2 没有对应支付left join 应输出 null).assignTimestampsAndWatermarks(WatermarkStrategy.PaymentforBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((e,t)-e.timestamp));DataStreamJoinResultresultorderStream.keyBy(o-o.orderId).connect(paymentStream.keyBy(p-p.orderId)).process(newLeftJoinFunction());result.print();env.execute();}}上面是实现Left Join的简单Demo在生产环境中建议给状态设置TTL并使用RocksDB state backend。选型场景推荐方式简单等值 join时间对齐Window Join有时间偏差的 inner joinInterval Join需要 outer join 或复杂逻辑CoProcessFunctionFlink SQL如果使用Table API或者SQL支持更丰富的语义类型说明Regular Join全量 join状态持续保留Interval Join基于时间区间Temporal Join流与版本表/维表 joinLookup Join流关联外部维表如 MySQL、HBaseRegular Join全量 join两侧数据都会永久保留在 state 中适合数据量不大的场景。-- inner joinSELECTo.order_id,p.amountFROMorders oJOINpayments pONo.order_idp.order_id-- left joinSELECTo.order_id,p.amountFROMorders oLEFTJOINpayments pONo.order_idp.order_id-- full outer joinSELECTo.order_id,p.amountFROMorders oFULLOUTERJOINpayments pONo.order_idp.order_idRegular Join的特点state 无限增长生产环境需要配置 state TTLtable.exec.state.ttl:86400000# 1天单位毫秒Interval Join基于事件时间区间的 joinstate 可以自动清理比 Regular Join 更适合生产。SELECTo.order_id,p.amountFROMorders o,payments pWHEREo.order_idp.order_idANDp.pay_timeBETWEENo.order_time-INTERVAL5MINUTEANDo.order_timeINTERVAL10MINUTEInterval Join的特点只支持 inner join两张表都必须有事件时间属性区间范围要合理太大会导致 state 积压Temporal Join时态表Join流关联一张有版本概念的表获取事件发生时刻对应的快照数据典型场景是汇率、价格表。-- 定义版本表有主键 事件时间CREATETABLEexchange_rates(currency STRING,rateDOUBLE,update_timeTIMESTAMP(3),WATERMARKFORupdate_timeASupdate_time-INTERVAL5SECOND,PRIMARYKEY(currency)NOTENFORCED)WITH(...);-- 流关联版本表取事件发生时的汇率SELECTo.order_id,o.amount*r.rateASamount_usdFROMorders oLEFTJOINexchange_ratesFORSYSTEM_TIMEASOFo.order_timeASrONo.currencyr.currencyTemporal Join的特点获取的是事件时间对应的历史快照不是最新值state 可以自动清理Lookup Join流关联外部维表MySQL、HBase、Redis 等实时查询外部系统补充字段。-- 定义外部维表CREATETABLEuser_info(user_id STRING,user_name STRING,ageINT,PRIMARYKEY(user_id)NOTENFORCED)WITH(connectorjdbc,urljdbc:mysql://localhost:3306/db,table-nameuser_info);-- 使用处理时间做 lookup joinSELECTo.order_id,u.user_name,o.amountFROMorders oLEFTJOINuser_infoFORSYSTEM_TIMEASOFo.proc_timeASuONo.user_idu.user_idLookup Join的特点只支持处理时间Processing Time每条左流数据触发一次外部查询如果不想每条左流数据触发一次外部查询可以使用缓存机制但对于变动频繁的维表不合适。CREATETABLEuser_info(user_id STRING,user_name STRING)WITH(connectorjdbc,urljdbc:mysql://localhost:3306/db,table-nameuser_info,-- 开启缓存lookup.cachePARTIAL,-- 最大缓存行数lookup.partial-cache.max-rows10000,-- 写入后过期时间lookup.partial-cache.expire-after-write60s,-- 读取后过期时间lookup.partial-cache.expire-after-access30s,-- 是否缓存空值lookup.partial-cache.cache-missing-keytrue);Window Join基于窗口的 join窗口内数据匹配窗口结束后 state 自动清理。SELECTo.order_id,p.amountFROMTABLE(TUMBLE(TABLEorders,DESCRIPTOR(order_time),INTERVAL10MINUTES))oJOINTABLE(TUMBLE(TABLEpayments,DESCRIPTOR(pay_time),INTERVAL10MINUTES))pONo.order_idp.order_idANDo.window_startp.window_startANDo.window_endp.window_endWindow Join的特点支持 Tumbling / Hopping / Cumulate 三种窗口。只支持 inner join。选型类型outer joinstate 自动清理适用场景Regular Join✅❌数据量小逻辑简单Interval Join❌✅有时间关联的双流Temporal Join✅✅关联历史版本快照Lookup Join✅✅关联外部维表Window Join❌✅窗口内聚合后关联适用场景1. 订单 支付流订单产生后等待支付事件到来进行关联判断是否支付成功。时间有偏差适合 Interval Join 或 CoProcessFunction。2. 行为流 用户信息流用户点击/浏览行为流关联用户画像流做实时enrichment。用户信息变化不频繁适合 Temporal Join 或 Lookup Join。3. 广告曝光 点击流曝光事件和点击事件分属两条流需要关联计算点击率。时间窗口内匹配适合 Window Join 或 Interval Join。4. 日志流 告警规则流实时日志流关联动态告警规则流规则会实时更新。适合 CoProcessFunction规则侧用 broadcast state。5. 交易流 风控事件流交易发生后关联风控系统产生的风险事件判断是否拦截。对延迟敏感适合 Interval Join。6. 多源数据合并同一业务实体的数据来自不同系统如 MySQL binlog Kafka 业务流需要合并成完整记录。适合 CoProcessFunction 自定义 state 管理。7. 实时对账银行/支付场景两条流分别来自不同系统的流水记录需要实时核对是否一致。适合 CoProcessFunction 实现 full outer join。
Flink知识点（一）｜Flink中的双流关联
发布时间：2026/7/14 7:50:22
在FLink中经常会遇到双流关联的的场景比如订单流和支付流的关联用户行为流和用户信息流等。那么在Flink中都提供了哪些双流的关联呢下面主要从DataStream和Flink SQL角度入手说明。DataStreamWindow Join窗口Join两条流在同一个时间窗口内进行Join窗口结束时出触发计算。// 设置10秒的窗口期stream1.join(stream2).where(r-r.id).equalTo(r-r.id).window(TumblingEventTimeWindows.of(Time.seconds(10))).apply((e1,e2)-e1 e2);Window Join的特点只能做 inner join两条流的数据必须落在同一个窗口内才能匹配支持 Tumbling / Sliding / Session WindowWindow Join只能做 inner joinjoin不上的数据会直接丢弃没有机制输出到侧输出流所以谨慎使用。Interval Join区间Join基于事件时间允许一条流的数据在另一条流数据的时间范围内进行匹配。// stream1流的一个记录可以在stream2前后5秒的这个区间匹配stream1.keyBy(r-r.id).intervalJoin(stream2.keyBy(r-r.id)).between(Time.seconds(-5),Time.seconds(5)).process(newProcessJoinFunction(){OverridepublicvoidprocessElement(Leftl,Rightr,Contextctx,CollectorStringout){out.collect(l r);}});Interval Join的特点只支持 inner join只支持 event time比 Window Join 更灵活适合时间偏差场景同样Interval Join只能做 inner joinjoin不上的数据会直接丢弃没有机制输出到侧输出流所以谨慎使用。CoProcessFunction底层 APICoprocessFunction是最灵活的方式可以实现inner / left / right / full outer join以及各种自定义逻辑。stream1.keyBy(r-r.id).connect(stream2.keyBy(r-r.id)).process(newCoProcessFunctionLeft,Right,String(){// 用 State 缓存两侧数据自行实现匹配逻辑MapStateString,LeftleftState;MapStateString,RightrightState;OverridepublicvoidprocessElement1(Leftl,Contextctx,CollectorStringout){// 查 rightState匹配则输出否则存入 leftState}OverridepublicvoidprocessElement2(Rightr,Contextctx,CollectorStringout){// 查 leftState匹配则输出否则存入 rightState}});CoProcessFunction的特点完全自定义支持所有 join 类型需要手动管理 State 和 TTL防止状态无限增长适合复杂业务逻辑使用CoProcessFunction实现Left Join下面使用CoProcessFunction实现Left Join。publicclassLeftJoinExample{// 左流数据DataAllArgsConstructorpublicstaticclassOrder{publicStringorderId;publicStringuserId;publiclongtimestamp;}// 右流数据DataAllArgsConstructorpublicstaticclassPayment{publicStringorderId;publicdoubleamount;publiclongtimestamp;}// 输出结果DataAllArgsConstructorpublicstaticclassJoinResult{publicStringorderId;publicStringuserId;publicDoubleamount;// 右流未匹配时为 null}publicstaticclassLeftJoinFunctionextendsCoProcessFunctionOrder,Payment,JoinResult{// 未匹配的左流数据privateMapStateString,OrderorderState;// 右流数据缓存privateMapStateString,PaymentpaymentState;// 等待匹配的超时时间privatestaticfinallongJOIN_TIMEOUT10_000L;Overridepublicvoidopen(Configurationparameters){orderStategetRuntimeContext().getMapState(newMapStateDescriptor(order-state,String.class,Order.class));paymentStategetRuntimeContext().getMapState(newMapStateDescriptor(payment-state,String.class,Payment.class));}OverridepublicvoidprocessElement1(Orderorder,Contextctx,CollectorJoinResultout)throwsException{PaymentpaymentpaymentState.get(order.orderId);if(payment!null){// 右流已有数据直接 join 输出out.collect(newJoinResult(order.orderId,order.userId,payment.amount));paymentState.remove(order.orderId);}else{// 右流暂无数据缓存左流注册超时定时器orderState.put(order.orderId,order);ctx.timerService().registerEventTimeTimer(order.timestampJOIN_TIMEOUT);}}OverridepublicvoidprocessElement2(Paymentpayment,Contextctx,CollectorJoinResultout)throwsException{OrderorderorderState.get(payment.orderId);if(order!null){// 左流已有数据直接 join 输出out.collect(newJoinResult(order.orderId,order.userId,payment.amount));orderState.remove(payment.orderId);}else{// 左流暂无数据缓存右流等待paymentState.put(payment.orderId,payment);}}OverridepublicvoidonTimer(longtimestamp,OnTimerContextctx,CollectorJoinResultout)throwsException{// 超时触发左流数据仍未匹配输出 null 右侧left join 语义for(Map.EntryString,Orderentry:orderState.entries()){Orderorderentry.getValue();if(order.timestampJOIN_TIMEOUTtimestamp){out.collect(newJoinResult(order.orderId,order.userId,null));orderState.remove(entry.getKey());}}}}publicstaticvoidmain(String[]args)throwsException{StreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();DataStreamOrderorderStreamenv.fromElements(newOrder(o1,u1,1000L),newOrder(o2,u2,2000L)).assignTimestampsAndWatermarks(WatermarkStrategy.OrderforBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((e,t)-e.timestamp));DataStreamPaymentpaymentStreamenv.fromElements(newPayment(o1,99.9,3000L)// o2 没有对应支付left join 应输出 null).assignTimestampsAndWatermarks(WatermarkStrategy.PaymentforBoundedOutOfOrderness(Duration.ofSeconds(5)).withTimestampAssigner((e,t)-e.timestamp));DataStreamJoinResultresultorderStream.keyBy(o-o.orderId).connect(paymentStream.keyBy(p-p.orderId)).process(newLeftJoinFunction());result.print();env.execute();}}上面是实现Left Join的简单Demo在生产环境中建议给状态设置TTL并使用RocksDB state backend。选型场景推荐方式简单等值 join时间对齐Window Join有时间偏差的 inner joinInterval Join需要 outer join 或复杂逻辑CoProcessFunctionFlink SQL如果使用Table API或者SQL支持更丰富的语义类型说明Regular Join全量 join状态持续保留Interval Join基于时间区间Temporal Join流与版本表/维表 joinLookup Join流关联外部维表如 MySQL、HBaseRegular Join全量 join两侧数据都会永久保留在 state 中适合数据量不大的场景。-- inner joinSELECTo.order_id,p.amountFROMorders oJOINpayments pONo.order_idp.order_id-- left joinSELECTo.order_id,p.amountFROMorders oLEFTJOINpayments pONo.order_idp.order_id-- full outer joinSELECTo.order_id,p.amountFROMorders oFULLOUTERJOINpayments pONo.order_idp.order_idRegular Join的特点state 无限增长生产环境需要配置 state TTLtable.exec.state.ttl:86400000# 1天单位毫秒Interval Join基于事件时间区间的 joinstate 可以自动清理比 Regular Join 更适合生产。SELECTo.order_id,p.amountFROMorders o,payments pWHEREo.order_idp.order_idANDp.pay_timeBETWEENo.order_time-INTERVAL5MINUTEANDo.order_timeINTERVAL10MINUTEInterval Join的特点只支持 inner join两张表都必须有事件时间属性区间范围要合理太大会导致 state 积压Temporal Join时态表Join流关联一张有版本概念的表获取事件发生时刻对应的快照数据典型场景是汇率、价格表。-- 定义版本表有主键 事件时间CREATETABLEexchange_rates(currency STRING,rateDOUBLE,update_timeTIMESTAMP(3),WATERMARKFORupdate_timeASupdate_time-INTERVAL5SECOND,PRIMARYKEY(currency)NOTENFORCED)WITH(...);-- 流关联版本表取事件发生时的汇率SELECTo.order_id,o.amount*r.rateASamount_usdFROMorders oLEFTJOINexchange_ratesFORSYSTEM_TIMEASOFo.order_timeASrONo.currencyr.currencyTemporal Join的特点获取的是事件时间对应的历史快照不是最新值state 可以自动清理Lookup Join流关联外部维表MySQL、HBase、Redis 等实时查询外部系统补充字段。-- 定义外部维表CREATETABLEuser_info(user_id STRING,user_name STRING,ageINT,PRIMARYKEY(user_id)NOTENFORCED)WITH(connectorjdbc,urljdbc:mysql://localhost:3306/db,table-nameuser_info);-- 使用处理时间做 lookup joinSELECTo.order_id,u.user_name,o.amountFROMorders oLEFTJOINuser_infoFORSYSTEM_TIMEASOFo.proc_timeASuONo.user_idu.user_idLookup Join的特点只支持处理时间Processing Time每条左流数据触发一次外部查询如果不想每条左流数据触发一次外部查询可以使用缓存机制但对于变动频繁的维表不合适。CREATETABLEuser_info(user_id STRING,user_name STRING)WITH(connectorjdbc,urljdbc:mysql://localhost:3306/db,table-nameuser_info,-- 开启缓存lookup.cachePARTIAL,-- 最大缓存行数lookup.partial-cache.max-rows10000,-- 写入后过期时间lookup.partial-cache.expire-after-write60s,-- 读取后过期时间lookup.partial-cache.expire-after-access30s,-- 是否缓存空值lookup.partial-cache.cache-missing-keytrue);Window Join基于窗口的 join窗口内数据匹配窗口结束后 state 自动清理。SELECTo.order_id,p.amountFROMTABLE(TUMBLE(TABLEorders,DESCRIPTOR(order_time),INTERVAL10MINUTES))oJOINTABLE(TUMBLE(TABLEpayments,DESCRIPTOR(pay_time),INTERVAL10MINUTES))pONo.order_idp.order_idANDo.window_startp.window_startANDo.window_endp.window_endWindow Join的特点支持 Tumbling / Hopping / Cumulate 三种窗口。只支持 inner join。选型类型outer joinstate 自动清理适用场景Regular Join✅❌数据量小逻辑简单Interval Join❌✅有时间关联的双流Temporal Join✅✅关联历史版本快照Lookup Join✅✅关联外部维表Window Join❌✅窗口内聚合后关联适用场景1. 订单 支付流订单产生后等待支付事件到来进行关联判断是否支付成功。时间有偏差适合 Interval Join 或 CoProcessFunction。2. 行为流 用户信息流用户点击/浏览行为流关联用户画像流做实时enrichment。用户信息变化不频繁适合 Temporal Join 或 Lookup Join。3. 广告曝光 点击流曝光事件和点击事件分属两条流需要关联计算点击率。时间窗口内匹配适合 Window Join 或 Interval Join。4. 日志流 告警规则流实时日志流关联动态告警规则流规则会实时更新。适合 CoProcessFunction规则侧用 broadcast state。5. 交易流 风控事件流交易发生后关联风控系统产生的风险事件判断是否拦截。对延迟敏感适合 Interval Join。6. 多源数据合并同一业务实体的数据来自不同系统如 MySQL binlog Kafka 业务流需要合并成完整记录。适合 CoProcessFunction 自定义 state 管理。7. 实时对账银行/支付场景两条流分别来自不同系统的流水记录需要实时核对是否一致。适合 CoProcessFunction 实现 full outer join。
)
