📚 课程目录第2章 Flink State与Join基础

Flink State 与 Join 基础

好，咱们正式开始第二讲。上一章我聊了聊风控场景对低延迟的执念，那这一章就得把地基打牢——Flink 的 State 和 Join 到底是怎么玩的？

说实话，我见过不少同学，代码写得飞起，但一问到 State 的底层存储结构就含糊了。结果呢？线上 OOM 了都不知道怎么死的。所以这一章，咱们把 KeyedState 的几个主力类型掰开揉碎，再把 Flink 那三个 Join 算子理清楚。

Flink KeyedState 简介

先问个问题：为什么风控场景里，State 这么重要？

你想想看，风控规则往往是跨事件的。比如「5 分钟内同一设备登录超过 3 次就告警」。这个「5 分钟内」和「同一设备」的信息，必须有个地方存着。Flink 的 State 就是干这个的。

KeyedState，顾名思义，是按 key 分片的 State。每个 key 都有自己的状态视图。在风控里，key 通常是 deviceId、userId、ip 这些维度。

核心要点：KeyedState 只存在于 KeyedStream 上。你得先 .keyBy()，才能用 State。

我个人习惯把 KeyedState 理解成一个「分布式 HashMap」。每个并行子任务管理自己那部分 key 的状态。Flink 负责把状态做 checkpoint，保证故障恢复。

ValueState / MapState / ListState

这三种是最常用的。咱们一个一个说。

ValueState

最简单的状态类型。存一个值，更新一个值。适合计数器、标记位、最近一次事件时间。

// 风控场景：记录用户最近一次登录时间
ValueState<Long> lastLoginTime = getRuntimeContext()
    .getState(new ValueStateDescriptor<>("last-login", Types.LONG));

// 更新
lastLoginTime.update(System.currentTimeMillis());

// 读取
Long lastTime = lastLoginTime.value();

我在项目中遇到过一个问题：ValueState 默认的 TTL 没设置，结果状态越积越多，checkpoint 越来越大。后来统一加了 StateTtlConfig，才控制住。

避坑指南：ValueState 的 value() 返回 null 表示从未写入过。别用 0 或 -1 做默认值，直接用 null 判断更安全。

MapState

存 key-value 对。适合需要按子维度聚合的场景。比如「每个 IP 关联的设备列表」。

MapState<String, Integer> deviceCountMap = getRuntimeContext()
    .getMapState(new MapStateDescriptor<>("device-count", Types.STRING, Types.INT));

// 累加设备出现次数
Integer count = deviceCountMap.get(deviceId);
if (count == null) {
    deviceCountMap.put(deviceId, 1);
} else {
    deviceCountMap.put(deviceId, count + 1);
}

MapState 比 ValueState 灵活，但要注意：遍历所有条目时，数据量大会有性能问题。我曾经在线上看到一个任务，MapState 里存了上百万个 key，每次遍历都卡好几秒。后来改成了按需查询，问题解决。

ListState

存一个有序列表。适合保存事件序列，比如「最近 10 次登录记录」。

ListState<LoginEvent> recentLogins = getRuntimeContext()
    .getListState(new ListStateDescriptor<>("recent-logins", LoginEvent.class));

// 添加事件
recentLogins.add(event);

// 遍历
for (LoginEvent e : recentLogins.get()) {
    // 处理
}

ListState 有个坑：它不保证元素去重，也不保证顺序（除非你自己控制）。我建议只在需要「追加」语义时用它，否则优先考虑 MapState。

状态类型	适用场景	注意事项
ValueState	计数器、标记位、单值缓存	注意 TTL 设置，避免状态膨胀
MapState	维度聚合、字典映射	避免全量遍历，按 key 查询
ListState	事件序列、日志缓存	不保证去重和顺序

Flink Join 算子概览

风控里最头疼的就是 Join。两个流要关联，延迟还不能高。Flink 给了三个主力算子，各有各的脾气。

Window Join

最直观的 Join 方式。把两个流放在同一个时间窗口里，按 key 做关联。

stream1.join(stream2)
    .where(e1 -> e1.getKey())
    .equalTo(e2 -> e2.getKey())
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .apply((e1, e2) -> { /* 合并逻辑 */ });

Window Join 的优点是简单。缺点也很明显：必须等窗口结束才能输出结果。在风控场景里，5 秒的延迟有时候都嫌长。

注意：Window Join 默认是 inner join。如果某个流在窗口内没数据，就不会输出。风控里经常需要 left join 或 anti join，Window Join 做不到。

Interval Join

这个算子就灵活多了。它允许两个流的事件在「时间区间」内关联，不用等窗口结束。

stream1.keyBy(e -> e.getKey())
    .intervalJoin(stream2.keyBy(e -> e.getKey()))
    .between(Time.seconds(-5), Time.seconds(5))
    .process(new ProcessJoinFunction<>() {
        @Override
        public void processElement(Object left, Object right, Context ctx, Collector<Object> out) {
            // 实时输出
        }
    });

Interval Join 的延迟更低——事件一到，只要匹配到另一条流的数据，立刻输出。我在做实时反欺诈时，用的就是 Interval Join 来关联登录事件和交易事件。

但要注意：Interval Join 只支持 event time。如果你用 processing time，时间边界会乱掉。

Temporal Join

这个算子比较特殊。它用于「流」关联「版本化表」。比如实时关联用户画像、设备指纹这类变化缓慢的数据。

// 假设 userProfile 是一个 TemporalTable
stream.keyBy(e -> e.getUserId())
    .join(userProfile)
    .where(e -> e.getUserId())
    .equalTo(p -> p.getUserId())
    .window(TumblingEventTimeWindows.of(Time.seconds(1)))
    .apply((e, p) -> { /* 合并 */ });

Temporal Join 的好处是：能拿到事件发生时，那个时刻的版本数据。而不是当前的最新数据。这在风控审计里特别重要——你得知道当时用户画像是什么，而不是现在的。

我的建议：如果关联的数据变化频率低（小时级），用 Temporal Join。如果变化快（秒级），用 Interval Join 更合适。

嗯，这张图把三个算子的核心差异都标出来了。风控场景里，我首选 Interval Join，因为它延迟最低。Temporal Join 适合关联静态数据。Window Join 嘛……除非你对延迟不敏感，否则慎用。