为什么需要流 Join
实时分析经常需要关联两条或多条流：
- 订单流 + 支付流 → 订单支付完成率
- 点击流 + 购买流 → 转化率分析
- 事件流 + 规则流 → 动态告警
eventd v3 的场景：把用户订单事件和用户资料流关联起来，为订单补齐用户等级、地区等信息。
Interval Join
Interval Join 在两条流的事件时间窗口内匹配：
订单流 支付流
(orderId=1, t=0) → (orderId=1, t=5) ✓ 匹配（支付在订单后 5 秒内）
(orderId=2, t=10) → (orderId=2, t=60) ✗ 不匹配（超过 30 秒窗口）
// 本章各段 Join 共用这组模型。字段里的 timestamp 是业务时间字段，
// 与 assignTimestampsAndWatermarks 分配的事件时间戳一致，
// 但 Join 窗口只认后者，业务字段不参与匹配
record OrderEvent(String orderId, String userId, double amount, long timestamp) {}
record PaymentEvent(String orderId, String status, long timestamp) {}
// Interval Join 的输出：订单 + 支付
record PaidOrder(
String orderId,
String userId,
double amount,
String payStatus,
long payDelayMs
) {}
// CoProcessFunction 的输出：订单 + 用户资料
// orderTime 必须一起带下去，否则第二次 Join 算不出支付延迟
record OrderWithUser(
String orderId,
String userId,
double amount,
long orderTime,
String tier, // 资料未到时为 null
String region // 资料未到时为 null
) {}
// 两次 Join 的最终结果
record EnrichedOrder(
String orderId,
String userId,
double amount,
String tier,
String region,
String payStatus,
long payDelayMs
) {}
// 订单流
DataStream<OrderEvent> orders = env.fromSource(orderSource, watermarkStrategy, "orders");
// 支付流
DataStream<PaymentEvent> payments = env.fromSource(paymentSource, watermarkStrategy, "payments");
// Interval Join：订单后 [0, 30s] 内的支付
orders.keyBy(OrderEvent::orderId)
.intervalJoin(payments.keyBy(PaymentEvent::orderId))
.between(Duration.ZERO, Duration.ofSeconds(30))
.process(new ProcessJoinFunction<OrderEvent, PaymentEvent, PaidOrder>() {
@Override
public void processElement(
OrderEvent order,
PaymentEvent payment,
Context ctx,
Collector<PaidOrder> out) {
out.collect(new PaidOrder(
order.orderId(),
order.userId(),
order.amount(),
payment.status(),
payment.timestamp() - order.timestamp() // 支付延迟
));
}
});
| 参数 | 含义 | 示例 |
|---|---|---|
between(lower, upper) | 右侧元素的时间必须满足 right.ts - left.ts ∈ [lower, upper] | between(0, 30s)：支付不早于订单、且不晚于订单后 30 秒 |
keyBy | Join key，两侧必须同 key | 按 orderId |
lower 为负值时允许右流先到。例如 between(-Duration.ofSeconds(5), Duration.ofSeconds(30)) 表示支付可以在订单前 5 秒到订单后 30 秒之间。
Interval Join 的限制：
- 只能 Join 两条流
- 必须有共同的 key
- 必须有事件时间和 Watermark
CoProcessFunction
更灵活的 Join 方式：两条流进入同一个算子，按各自事件处理：
// 两流连接
ConnectedStreams<OrderEvent, UserProfile> connected =
orders.keyBy(OrderEvent::userId)
.connect(userProfiles.keyBy(UserProfile::userId));
DataStream<OrderWithUser> enriched = connected.process(
new CoProcessFunction<OrderEvent, UserProfile, OrderWithUser>() {
private transient ValueState<UserProfile> profileState;
@Override
public void open(OpenContext parameters) throws Exception {
profileState = getRuntimeContext().getState(
new ValueStateDescriptor<>("profile", UserProfile.class));
}
@Override
public void processElement1(OrderEvent order, Context ctx,
Collector<OrderWithUser> out) throws Exception {
UserProfile profile = profileState.value();
if (profile != null) {
out.collect(new OrderWithUser(order.orderId(), order.userId(),
order.amount(), order.timestamp(),
profile.tier(), profile.region()));
} else {
// 用户信息未到。这里只标记为未补全，不缓存订单：
// 要缓存需再开一个 ValueState<List<OrderEvent>>，并用定时器兜底清理，
// 否则永不到达的订单会一直留在状态里
out.collect(new OrderWithUser(order.orderId(), order.userId(),
order.amount(), order.timestamp(), null, null));
}
}
@Override
public void processElement2(UserProfile profile, Context ctx,
Collector<OrderWithUser> out) throws Exception {
profileState.update(profile); // 保存最新用户资料
}
});
适用场景：
- 一个流更新频率低（如用户资料），另一个流高频（如订单）
- 不需要精确时间窗口匹配
- 需要自定义缓存策略
BroadcastStream 关联
用广播流实现动态规则关联：
// 主数据流：事件流
DataStream<Event> eventStream = env.fromSource(eventSource, watermarkStrategy, "events");
// 广播流：实时更新的规则
DataStream<Rule> ruleStream = env.fromSource(ruleSource, watermarkStrategy, "rules");
// 与 5.2 的 Rule 同名但口径不同：5.2 面向 4.x~8.x 的 POJO Event（getType()），
// 这里用 9 章之后的 v3 record Event（type()/amount()）。同一个作业内只保留一份。
record Rule(String type, String action, double threshold) {
boolean matches(Event event) {
return type.equals(event.type()) && event.amount() > threshold;
}
}
record Alert(Event event, Rule rule, long triggeredAt) {}
MapStateDescriptor<String, Rule> ruleStateDescriptor =
new MapStateDescriptor<>("rules", String.class, Rule.class);
BroadcastStream<Rule> broadcastRules = ruleStream.broadcast(ruleStateDescriptor);
// 事件 + 规则关联
eventStream
.keyBy(Event::type)
.connect(broadcastRules)
.process(new KeyedBroadcastProcessFunction<String, Event, Rule, Alert>() {
@Override
public void processElement(Event event, ReadOnlyContext ctx,
Collector<Alert> out) throws Exception {
Rule rule = ctx.getBroadcastState(ruleStateDescriptor).get(event.type());
if (rule != null && rule.matches(event)) {
out.collect(new Alert(event, rule, ctx.timestamp()));
}
}
@Override
public void processBroadcastElement(Rule rule, Context ctx,
Collector<Alert> out) throws Exception {
ctx.getBroadcastState(ruleStateDescriptor).put(rule.type(), rule);
}
});
eventd v3 的风控规则使用 BroadcastStream：规则从后台管理页面下发到 Kafka Topic → Flink 作业实时更新，无需重启作业。
eventd v3 多维关联
eventd v3 的关联场景：
订单流 (userId, amount, symbol, eventTime)
+ 用户资料流 (userId, tier, region, vip_since)
= 增强订单 (userId, amount, symbol, tier, region)
订单流
+ 支付流 (orderId, status, payTime)
= 支付完成订单 (orderId, userId, amount, status, payDelayMs)
// 第一个 Join：订单 + 用户资料（CoProcessFunction）
DataStream<OrderWithUser> enrichedByUser = orders
.keyBy(OrderEvent::userId)
.connect(profiles.keyBy(UserProfile::userId))
.process(new OrderEnrichFunction());
// 第二个 Join：增强订单 + 支付（Interval Join）
enrichedByUser.keyBy(OrderWithUser::orderId)
.intervalJoin(payments.keyBy(PaymentEvent::orderId))
.between(Duration.ZERO, Duration.ofMinutes(5))
.process(new PaymentJoinFunction());
static class PaymentJoinFunction
extends ProcessJoinFunction<OrderWithUser, PaymentEvent, EnrichedOrder> {
@Override
public void processElement(OrderWithUser order, PaymentEvent payment,
Context ctx, Collector<EnrichedOrder> out) {
out.collect(new EnrichedOrder(
order.orderId(), order.userId(), order.amount(),
order.tier(), order.region(),
payment.status(),
payment.timestamp() - order.orderTime()));
}
}
OrderEnrichFunction 就是把上一节的匿名 CoProcessFunction 抽成具名类。
注意两次 Join 用的 key 不同（先按 userId，后按 orderId）：key 变了就触发一次 keyBy 的 shuffle，状态也按新 key 重新分区。CoProcessFunction 里的 profileState 在第一步之前按 userId 分区，第二步不会带着它走，所以第二次 Join 只能拿到已经写进 OrderWithUser 的字段。
Stream Join 选型指南
| Join 类型 | 适用 | 限制 |
|---|---|---|
intervalJoin | 两条流有时间窗口约束的 Join | 只能 2 条流，必须有 key |
coGroup | 两条流在窗口内关联 | 必须用窗口（Tumbling/Sliding） |
CoProcessFunction | 完全自定义 Join 逻辑 | 需要自己管理状态 |
BroadcastStream | 一多一少（动态规则） | 每个并行实例各存一份相同副本，State 大小乘以并行度；且状态修改只能在广播侧进行 |
小结
| Join 方式 | 灵活度 | 状态 | 适用 |
|---|---|---|---|
| Interval Join | 低 | 自动窗口状态 | 有时间窗口的两流关联 |
| Broadcast Join | 中 | BroadcastState | 动态规则 |
| CoProcessFunction | 高 | 自定义 ValueState | 完全自定义关联逻辑 |
下一节 11.3 状态 TTL 与过期清理 ：StateTtlConfig、过期策略与清理。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。