阅读本文需要提前了解 connect 和 ProcessFunction 相关的知识。如果不了解的同学可以先通过官网或其他资料熟悉一下。
一、案例分析
在生产环境中，我们经常会遇到双流匹配的案例，例如：
- 一个订单包含了订单主体信息和商品的信息。
- 外卖行业，一个订单包含了订单付款信息和派送信息。
- 互联网广告行业，一次点击包含了用户的点击行为日志和计费日志。
- 等其他相关的案例
上述这些案例都需要涉及到双流匹配的操作，也就是所谓的双流 join。下面用一个案例来详解如何用 connect 实现双流 join。
本文案例
一个订单分成了大订单和小订单，大小订单对应的数据流来自 Kafka 不同的 Topic，需要在两个数据流中按照订单 Id 进行匹配，这里认为相同订单 id 的两个流的延迟最大为 60s。大订单和小订单匹配成功后向下游发送，若 60s 还未匹配成功，意味着当前只有一个流来临，则认为订单异常，需要将数据进行侧流输出。
思路描述
提取两个流的时间戳，因为要通过订单 Id 进行匹配，所以这里按照订单 Id 进行 keyBy，然后两个流 connect，大订单和小订单的处理逻辑一样，两个流通过 ValueState 进行关联。假如大订单流对应的数据先来了，需要将大订单的相关信息保存到大订单的 ValueState 状态中，注册一个 60s 之后的定时器。
- 如果 60s 内来了小订单流对应的数据来了，则将两个数据拼接发送到下游。
- 如果 60s 内小订单流对应的数据还没来，就会触发 onTimer，然后进行侧流输出。
如果小订单流对应的数据先到，也是同样的处理逻辑，先将小订单的信息保存到小订单的 ValueState 中，注册 60s 之后的定时器。
二、实现
用代码来讲述如何实现，首先要配置 Checkpoint 等参数，这里就不详细阐述。
1. 定义订单类
这里大小订单都使用同一个类演示：
@Data
public class Order {
/** 订单发生的时间 */
long time;
/** 订单 id */
String orderId;
/** 用户id */
String userId;
/** 商品id */
int goodsId;
/** 价格 */
int price;
/** 城市 */
int cityId;
}
2. 从 Kafka 的 topic 读取大小订单数据
读取大订单数据，从 json 解析成 Order 类。从 Order 中提取 EventTime、并分配 WaterMark。按照订单 id 进行 keyBy 得到 bigOrderStream。
// 读取大订单数据，读取的是 json 类型的字符串
FlinkKafkaConsumerBase<String> consumerBigOrder =
new FlinkKafkaConsumer011<>("big_order_topic_name",
new SimpleStringSchema(),
KafkaConfigUtil.buildConsumerProps(KAFKA_CONSUMER_GROUP_ID))
.setStartFromGroupOffsets();
KeyedStream<Order, String> bigOrderStream = env.addSource(consumerBigOrder)
// 有状态算子一定要配置 uid
.uid(KAFKA_TOPIC)
// 过滤掉 null 数据
.filter(Objects::nonNull)
// 将 json 解析为 Order 类
.map(str -> JSON.parseObject(str, Order.class))
// 提取 EventTime，分配 WaterMark
.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<Order>
(Time.seconds(60)) {
@Override
public long extractTimestamp(Order order) {
return order.getTime();
}
})
// 按照 订单id 进行 keyBy
.keyBy(Order::getOrderId);
小订单的处理逻辑与上述流程完全类似，只不过读取的 topic 不是同一个。
// 小订单处理逻辑与大订单完全一样
FlinkKafkaConsumerBase<String> consumerSmallOrder =
new FlinkKafkaConsumer011<>("small_order_topic_name",
new SimpleStringSchema(),
KafkaConfigUtil.buildConsumerProps(KAFKA_CONSUMER_GROUP_ID))
.setStartFromGroupOffsets();
KeyedStream<Order, String> smallOrderStream = env.addSource(consumerSmallOrder)
.uid(KAFKA_TOPIC)
.filter(Objects::nonNull)
.map(str -> JSON.parseObject(str, Order.class))
.assignTimestampsAndWatermarks(
new BoundedOutOfOrdernessTimestampExtractor<Order>
(Time.seconds(10)) {
@Override
public long extractTimestamp(Order order) {
return order.getTime();
}
})
.keyBy(Order::getOrderId);
3. connect 连接大小订单流，使用 process 进行匹配
再次描述一下处理流程：
两个流通过 ValueState 进行关联，假如大订单流对应的数据先来了，需要将大订单的相关信息保存到大订单的 ValueState 状态中，注册一个 60s 之后的定时器。
- 如果 60s 内来了小订单流对应的数据来了，则将两个数据拼接发送到下游。
- 如果 60s 内小订单流对应的数据还没来，就会触发 onTimer，然后进行侧流输出。
需要提前定义好侧流输出需要用到的 OutTag：
private static OutputTag<Order> bigOrderTag = new OutputTag<>("bigOrder");
private static OutputTag<Order> smallOrderTag = new OutputTag<>("smallOrder")
代码实现如下：
// 使用 connect 连接大小订单的流，然后使用 CoProcessFunction 进行数据匹配
SingleOutputStreamOperator<Tuple2<Order, Order>> resStream = bigOrderStream
.connect(smallOrderStream)
.process(new CoProcessFunction<Order, Order, Tuple2<Order, Order>>() {
// 大订单数据先来了，将大订单数据保存在 bigState 中。
ValueState<Order> bigState;
// 小订单数据先来了，将小订单数据保存在 smallState 中。
ValueState<Order> smallState;
// 大订单的处理逻辑
@Override
public void processElement1(Order bigOrder, Context ctx,
Collector<Tuple2<Order, Order>> out)
throws Exception {
// 获取当前 小订单的状态值
Order smallOrder = smallState.value();
// smallOrder 不为空表示小订单先来了，直接将大小订单拼接发送到下游
if (smallOrder != null) {
out.collect(Tuple2.of(smallOrder, bigOrder));
// 清空小订单对应的 State 信息
smallState.clear();
} else {
// 小订单还没来，将大订单放到状态中，并注册 1 分钟之后触发的 timerState
bigState.update(bigOrder);
// 1 分钟后触发定时器，当前的 eventTime + 60s
long time = bigOrder.getTime() + 60000;
ctx.timerService().registerEventTimeTimer(time);
}
}
@Override
public void processElement2(Order smallOrder, Context ctx,
Collector<Tuple2<Order, Order>> out)
throws Exception {
// 这里先省略代码，小订单的处理逻辑与大订单的处理逻辑完全类似
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<Tuple2<Order, Order>> out)
throws Exception {
// 定时器触发了，即 1 分钟内没有接收到两个流。
// 大订单不为空，则将大订单信息侧流输出
if (bigState.value() != null) {
ctx.output(bigOrderTag, bigState.value());
}
// 小订单不为空，则将小订单信息侧流输出
if (smallState.value() != null) {
ctx.output(smallOrderTag, smallState.value());
}
bigState.clear();
smallState.clear();
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// 初始化状态信息
bigState = getRuntimeContext().getState(
new ValueStateDescriptor<>("bigState", Order.class));
smallState = getRuntimeContext().getState(
new ValueStateDescriptor<>("smallState", Order.class));
}
});
小优化
假如 60s 以内，两个流的数据都到了，也就是执行了 out.collect(Tuple2.of(smallOrder, bigOrder)); 还有必要触发定时器吗？
定时器的目的是为了保证当 60s 时间到了，仍然有一个流还未到达。那么当两个流都到达时，没有必要再去触发定时器。所以当两个流都到达时，可以删除注册的定时器。（定时器的维护和触发也是需要成本的，所以及时清理这些垃圾是一个比较好的习惯）
改造后的代码如下，申请了一个 ValueState 类型的 timerState 用于维护注册的定时器时间，如果两个流都到达时，触发 delete 操作，同时要注意调用 timerState.clear() 去清理 timerState 的状态信息。
// 大订单的处理逻辑
@Override
public void processElement1(Order bigOrder, Context ctx,
Collector<Tuple2<Order, Order>> out)
throws Exception {
// 获取当前 小订单的状态值
Order smallOrder = smallState.value();
// smallOrder 不为空表示小订单先来了，直接将大小订单拼接发送到下游
if (smallOrder != null) {
out.collect(Tuple2.of(smallOrder, bigOrder));
// 清空小订单对应的 State 信息
smallState.clear();
// 这里可以将 Timer 清除。因为两个流都到了，没必要再触发 onTimer 了
ctx.timerService().deleteEventTimeTimer(timerState.value());
timerState.clear();
} else {
// 小订单还没来，将大订单放到状态中，并注册 1 分钟之后触发的 timerState
bigState.update(bigOrder);
// 1 分钟后触发定时器，并将定时器的触发时间保存在 timerState 中
long time = bigOrder.getTime() + 60000;
timerState.update(time);
ctx.timerService().registerEventTimeTimer(time);
}
}
4. 结果输出
这里直接将正常的输出结果还有策略输出都通过 print 进行输出，生产环境肯定是需要通过 Sink 输出到外部系统的。侧流输出的数据属于异常数据，需要保存到外部系统，进行特殊处理。
// 正常匹配到数据的 输出。生产环境肯定是需要通过 Sink 输出到外部系统
resStream.print();
// 只有大订单时，没有匹配到 小订单，属于异常数据，需要保存到外部系统，进行特殊处理
resStream.getSideOutput(bigOrderTag).print();
// 只有小订单时，没有匹配到 大订单，属于异常数据，需要保存到外部系统，进行特殊处理
resStream.getSideOutput(smallOrderTag).print();
env.execute(JOB_NAME);
三、 总结
Flink 使用 connect 实现双流 join 在 Flink 的 Streaming Api 中相对经常使用的 map、flatMap 等算子来讲已经属于比较复杂的场景了。文中开始处介绍的那些场景都可以通过本文的案例经过简单改造即可实现。而且 connect 实现双流 join 属于 Flink 面试的高频考点，希望读者通过本文有所收获。
本文详细代码请参考，感觉有用的同学帮忙点一波 Star：https://github.com/1996fanrui/fanrui-learning/blob/master/module-flink/src/main/java/com/dream/flink/connect/OrderMatch.java
近期还会更新几个实战案例，都是面试高频考点哦：
- Flink 使用 broadcast 实现维表或配置的实时更新
- Flink 实时计算分组 TopN 的问题，例如求每个城市最活跃的那 100 个用户
网友评论