企业官网建设流程全解析

这个需求我印象很深：某支付平台的风控同事跑来说，“能不能在我用户按下支付按钮到收到结果那几百毫秒里，判断出这笔交易是不是盗刷”。当晚要看到第一版结果。财务报表可以等第二天跑批，风控不行。这篇文章就用 Flink DataStream API 把整个实时欺诈检测作业从需求拆解、代码实现，到线上调优、上线验证的过程完整讲一遍，延续“实战一”里搭好的环境和基础概念，直接进入一个能跑、能报警、能调规则的真实项目。

适合刚写完 WordCount 想拿真实业务练手的开发者，也适合在风控团队做数据开发、想了解流计算到底怎么落地的同学。围绕 DataStream API 核心的 keyBy、状态、定时器、广播流，我会把每一步为什么这么写讲清楚，也会把线上遇到的坑原原本本列出来。看完你至少能独立搭出一个可以持续迭代的反欺诈实时管线。

1. 欺诈检测需求量化拆解：先把“可疑”变成可计算的规则

1.1 三个典型欺诈场景与判定条件

业务方给的需求往往只有一句话：“帮我做一个反欺诈任务。”真正动手前要问到底什么样算可疑。这个环节不做好，后面所有代码都是空中楼阁。我当时和风控团队开了三次需求会，把电子支付场景里最高频的三类风险行为定义成了可量化规则。

第一个是短时异地登录：同一账号在 1 小时内出现两个以上不同城市或不同设备，且相邻两次登录间隔小于 10 分钟。这条规则背后是典型的盗号场景，攻击者拿到密码后会在短时间内从不同 IP 尝试登录。第二个是高频小额试探：同一账号在 5 分钟内累计支付 3 笔以上，且每笔金额都低于 20 元。这很像是撞库成功后的小额转账试探，先拿小钱确认能不能支付再到大规模转走。第三个是单小时金额突变：账号当天累计消费金额超过用户近 7 日日均消费额的 3 倍以上，这类判断需要维护每个用户的历史行为基线。

这里有个关键细节：规则 1 里的“城市”不是用户自己上报的，而是从登录 IP 解析出来的归属地，由上游风控 SDK 解析后塞到事件字段里；“设备指纹”则由客户端上报的 deviceId 承担。也就是说，事件模型在源头就要把这两个字段带上。

1.2 规则的输出口径：命中、放行、降级

规则判定完不能只有“有风险”和“没风险”两种输出，真实风控系统需要分级处置。强风险规则，比如同一账号在 1 分钟内连续换了 3 台设备登录，直接输出高风险事件，支付环节要求二次验证；中风险规则，比如上面的高频小额试探，输出到人工审核队列，先冻结提现但不阻断支付；低风险规则，比如金额轻微超过阈值，只做监控和留存，不打扰用户。

分级输出看起来很简单，但代码上的差异很关键。分级意味着下游有好几套消费链路，告警事件要带上风险等级、命中规则 id、规则版本号。这些都应该是事件标准字段，而不是让人从告警正文里手动解析。这也是我反复跟业务强调的：普通表单事件可以是宽表，但风控告警事件必须按协议对齐，字段含义一点都不能含糊。

1.3 为什么这次选 DataStream API 而不是 Flink SQL

同一个需求用 Flink SQL 也能写，我用 SQL 试跑过一版，90% 的规则都可以用 OVER 窗口或者 GROUP BY 窗口表达。但最终还是把核心逻辑迁到了 DataStream API，原因有三个。

第一，多条规则要共享同一份状态。比如规则 1 要维护账号在 1 小时内的城市集合，规则 3 要维护近 7 天日均消费，如果用 SQL 写，不同窗口的中间结果要分开落表，再跨表 join，开发量和维护成本都上来了。DataStream 里一个 KeyedProcessFunction 可以在同一个 ValueState 里同时存放多套统计字段，代码集中，逻辑不容易散。

第二，风控规则的调整频率极高，阈值经常要按新攻击模式快速调整。DataStream API 配合广播流，可以把规则配置做成运行时可更新的状态，改阈值不用重启作业。Flink SQL 1.17 之后虽然也支持动态配置，但灵活度和社区资料成熟度远不如 DataStream 广播状态。

第三，部分规则存在跨事件才能判断的先后顺序关系，比如“首次登录城市 A、5 分钟后切到城市 B”，SQL 表达这类时序逻辑会很绕，DataStream 里用状态累积加定时器，读起来一目了然。再加上 Flink SQL 最终也会编译成 DataStream 作业，遇到性能瓶颈还是要回到底层 API 调优，不如一开始就把核心作业跑在 DataStream 上。

2. 核心作业的代码骨架：KeyedProcessFunction 完成状态累积与定时触发

2.1 事件模型与预处理

先定义一个统一的交易事件类 TransactionEvent，字段从上游消息队列的 JSON 直接映射过来，里面至少包含：accountId（账号）、eventId（全局唯一事件 id）、eventType（LOGIN 或 PAY）、city（IP 解析城市）、deviceId（设备指纹）、amount（金额）、ts（事件发生时间戳，毫秒）。

public class TransactionEvent { public String accountId; public String eventId; public String eventType; public String city; public String deviceId; public double amount; public long ts; }

拿到原始流之后不要直接进主逻辑，先做三层过滤。第一层过滤测试账号和内部标记数据，这些数据量不大但会污染统计基线；第二层过滤时间戳异常的数据，比如设备时钟跳变产生的时间戳是 1970 或 2050 年，不处理会让水印和定时器乱七八糟；第三层做 eventId 去重，上游在发生重试时会重复投递事件，我按 eventId 在状态里存了最近 5 分钟的 hash 集合，重复到达的事件直接丢弃。

我处理这个需求时，eventId 去重是上线后补上的。第一版没有去重，结果压测时因为消息中间件开启重试机制，同一笔支付事件被处理了两遍，高频小额试探规则直接误报。这个坑说明上游语义不总是 at-most-once，做流处理的人要在作业入口把幂等打好。

清洗完成后给事件分配时间戳和水印。这里选事件时间而不是处理时间，原因很直接：手机断网后重新联网，离线期间的行为会一起补传，如果按处理时间算，这些行为会被误判成“现在发生的一连串操作”，导致规则命中率失真。

DataStream<TransactionEvent> events = source .filter(new TestDataFilter()) .filter(new TimestampValidFilter()) .keyBy(e -> e.eventId) .process(new DeduplicateProcessFunction()) .assignTimestampsAndWatermarks( WatermarkStrategy.<TransactionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, timestamp) -> event.ts) );

2.2 keyBy 与状态设计

判定的核心维度是账号，所以主流程必须按 accountId 做 keyBy。这一步之后，同一个账号的所有事件都会路由到同一个并行子任务上，各任务只处理自己负责的那批账号，互不干扰。

接下来定义账号状态 AccountState，我把它放在 ValueState 里。需要注意，ValueState 不是让你持有一个无限膨胀的对象，比如“最近 1 小时事件列表”这种字段，如果存成 List 且不限制大小，账号行为密集时会直接把内存打爆。我的经验是：任何集合类字段都要设置上限，城市集合最多记 8 个，历史事件最多保留最近 20 条，窗口外的数据依赖定时器提前清理。

DataStream<AlertEvent> alertStream = events .keyBy(TransactionEvent::getAccountId) .process(new FraudDetectionProcessFunction());

2.3 定时器触发与状态滚动

核心逻辑放在自定义的 KeyedProcessFunction 里，方法名叫 FraudDetectionProcessFunction。processElement 里做三件事：更新账号统计状态；逐条检查规则，命中则通过侧输出把告警事件打出去；按事件时间注册每日滚动定时器。

状态滚动是这里最容易写错的地方。比如“当日累计金额”这个统计，需要在每天 0 点重置。如果只等到第二天第一条事件到达时才判断日期变了，这一天内如果一直没事件，定时器永远不会触发，数据就永远停留在昨天。正确做法是利用 TimerService 注册一个基于事件时间的定时器，在 0 点过 100 毫秒的位置触发 onTimer 回调，在回调里把当日累计金额清零，并注册下一天的定时器。

下面是简化后的 onTimer 逻辑：

@Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<AlertEvent> out) throws Exception { AccountState state = accountState.value(); if (state == null) { return; } LocalDateTime time = LocalDateTime.ofInstant( Instant.ofEpochMilli(timestamp), ZoneId.systemDefault()); if (time.getHour() == 0 && state.dayTotal != 0) { state.dayTotal = 0; state.alertSent = false; state.metricSnapshot = state.lastDayTotal; accountState.update(state); ctx.timerService().registerEventTimeTimer( timestamp + Duration.ofHours(1).toMillis()); } }

2.4 状态 TTL：防止资源泄漏

流作业会长期运行，key 是无上限的，每天有几十万新增账号。如果不给状态设置存活时间，那些一次登录后再也没有行为的账号会永远留在状态里，最终把 RocksDB 撑爆。

我给 AccountState 设置了 24 小时 TTL。这里有个经验：过期数据的可见性一定要设成 NeverReturnExpired，宁可多等一次清理，也不能让已经过期的账号数据继续参与规则判断，否则会产生一种很难排查的“幽灵误报”。

StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Duration.ofDays(1)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); ValueStateDescriptor<AccountState> descriptor = new ValueStateDescriptor<>("account_state", Types.POJO(AccountState.class)); descriptor.enableTimeToLive(ttlConfig);

3. 作业从测试环境到线上会碰到的四个“反直觉”问题

3.1 事件乱序对“首笔”判定的影响

测试环境用的是顺序数据，一切正常。一接线上真实流量，问题立刻出现：一个账号在 14:00 用设备 B 登录了一笔，这条日志因为手机断网在 14:07 才传到服务端，14:03 用设备 A 的登录记录反而先到。如果作业的逻辑是“看到新设备就报警”，这里就会误判成“换设备攻击”。

解决办法不是把水印延迟调大，而是调整规则判断方式。短时异地登录规则不再以“第一条事件是基准”，而是把 1 小时窗口内的城市、设备、时间全部累积在状态里，等到窗口闭合时统一判断。窗口内城市集合数量 >= 2，且相邻两次事件时间间隔小于 10 分钟，才触发告警。这样即使事件乱序，只要都在窗口内，集合统计不受影响。

3.2 并行度不是越大越好：状态分布与热点

欺诈检测作业按账号 keyBy，理论上加并行度就能水平扩展。但实际遇到一个棘手情况：有批“羊毛党”账号单日发起几千笔小额支付，这些账号的事件密度极高，会全部压在同一个子任务上。把这个子任务的并行度调大没有用，因为 keyBy 是按 key 哈希分区的，同一账号必须去同一个子任务。

我给这个账号级别的热点问题专门做了限流保护：在 processElement 入口处维护一个简单速率计数器，如果单个子任务每秒处理事件数超过阈值，直接向监控系统上报热点账号清单，并在告警事件里标记“疑似批量操作”，同时暂停对此账号的深度规则判断。流处理是连续不断的，用背压硬扛会拖慢整个作业，这里需要接受局部保护而不是全链路阻塞。

3.3 状态后端从内存换到 RocksDB

第一版图省事，状态后端用默认的 HashMap，跑测试没问题。预估线上账号规模后，我很快换成了 EmbeddedRocksDBStateBackend。原因是几百万账号的状态对象，每天光做 checkpoint 就要数秒，HashMap 后端一旦作业重启，恢复时间超长，而且大状态触发 full GC 的几率很高。

切换到 RocksDB 后有两个配套调整。一是打开增量 checkpoint，大状态作业全量 checkpoint 会因为数据量太大而超时，增量模式每次只提交变更部分，恢复速度明显变快。二是给 RocksDB 设置托管内存，让 Flink 统一管理堆外内存，避免和 JVM 堆内存互相争抢。

3.4 反压是常态，不是异常

告警事件如果直接在 process 函数里调用外部 HTTP 服务推送，一旦服务响应变慢，整个作业都会被拖住。我在生产环境遇到过一次审核服务 3 秒超时，结果所有 keyBy 子任务全部反压，端到端延迟从 50 毫秒飙到 5 秒。

后来把同步 HTTP 调用改成异步模式，用 AsyncFunction 配合线程池发起外部调用，同时把告警写入本地 Kafka 队列作为降级缓存。下游审核系统从 Kafka 消费告警，即使短时宕机也不会反压主流程。异步调用让 process 函数内部不等待外部响应，吞吐量比同步版本提升了近 4 倍。

4. 规则从硬编码到可配置：用广播状态实现热更新

4.1 为什么规则必须从代码中抽离

第一版上线后，风控团队几乎每周都要调一次阈值，比如把“单笔金额低于 20 元”改成 50 元，或者把“3 倍日均”改成 2.5 倍。每次调整都走代码发布，从提需求到上线至少要一天，攻击模式往往几个小时内就变化了。必须让规则变成运行时配置。

规则配置我设计成独立的 Rule 事件流。Rule 包含规则 id、规则类型、比较字段、比较阈值、时间窗口秒数、风险等级、规则版本号。每次风控平台修改规则，就向配置源写入一条新 Rule 事件，内容通过 Kafka 或配置中心推进流作业。

4.2 广播实现的完整步骤

Flink 处理动态配置的标准做法是 broadcast state。具体来说，把规则配置流 broadcast 到所有并行子任务，每个子任务维护一份完整的规则副本；普通事件流通过 connect 连接广播流，在处理交易事件时直接查本地规则。

MapStateDescriptor<String, Rule> ruleStateDescriptor = new MapStateDescriptor<>("online_rules", Types.STRING, Types.POJO(Rule.class)); BroadcastStream<Rule> ruleBroadcastStream = ruleSource.broadcast(ruleStateDescriptor); DataStream<AlertEvent> alertStream = events .keyBy(TransactionEvent::getAccountId) .connect(ruleBroadcastStream) .process(new FraudDetectionWithRuleProcessFunction());

在 FraudDetectionWithRuleProcessFunction 里，processBroadcastElement 负责接收最新规则并更新本地广播状态；processElement 负责读规则并逐条判断。改一条规则只需要往配置流里塞一条新事件，几秒钟后所有并行子任务都会更新，完全不用重启作业。

4.3 配置版本校验与灰度发布

规则可以热更新之后，紧接着出现一个事故隐患：某次把阈值从 3 倍调成了 2 倍，结果全平台告警量一夜暴涨，审核队列被塞满。教训是规则热更新必须带版本管理和灰度策略。

我在 Rule 事件里加了版本号字段，告警事件也记录 ruleVersion，这样后续回溯时能知道某条告警是哪个版本的规则判出来的。发布新规则时按流量灰度，可以先投放到测试账号，验证误报率正常后，再逐步放开到全流量。同时给高命中率规则配置冷却期：同一账号同一规则在 10 分钟内只触发一次告警，避免阈值放宽后同一批账号反复冲击下游审核系统。

5. 上线前的验证与压测：让作业体面地上线

5.1 用历史数据回放验证规则准确性

作业写完后不直接接生产流量，先用上周的历史日志做一轮全量回放。做法很简单：把消息中间件里上周的交易日志按原始时间顺序重新投递到一个新 topic，作业消费这个 topic，跑完后把所有告警事件和当时风控人工标记的结果做对比，算出精确率和召回率。

刚开始回放时，告警数量比预期高出差不多一倍。原因不是规则错了，而是历史日志里包含了大量接口调用失败后重试产生的重复事件，上一版作业还没做去重，导致“高频小额试探”规则被重复触发。这个验证环节帮我提前暴露了问题，而没有等到线上被用户投诉后再排查。每次调整规则阈值，我都建议用同一份历史数据重新跑一轮，输出一个像测试用例集一样的回归报告，防止改一条规则把另一条规则带偏。

5.2 压测配置与观测指标

上线前我用三档流量做了压测：正常流量 100 QPS、峰值流量 1000 QPS、极端流量 5000 QPS。每档跑 30 分钟，重点观测四个指标：端到端延迟 P99、checkpoint 时长、RocksDB 磁盘使用率、告警事件积压数量。

压测中发现的第一个问题是极端流量下 checkpoint 时长飙到 13 秒，接近 checkpoint 间隔，这会导致 pipeline 反复进入对齐状态，反而影响吞吐。我把 checkpoint 间隔从 30 秒调到 60 秒，并开启了 RocksDB 增量 checkpoint，极端流量下的同步耗时降到了 6 秒以内。第二个问题是告警事件在极端流量下有短暂积压，因为外部审核接口的处理能力有限，这正好验证了异步削峰和 Kafka 降级缓存设计的必要性。

5.3 告警降噪与人工复核闭环

规则判定完只是第一步，告警真正送到审核系统前必须降噪。我在主作业后面加了一层告警聚合算子，把规则命中事件按“规则 id + 账号”分组，10 分钟内只输出一条告警，后续命中信息以累加计数的方式附加到同一条告警上，也就是前面说的冷却期。实测下来，这个冷却期把审核队列的峰值压力降了 60% 左右，又不影响对风险账号的持续关注。

人工复核闭环也很重要。审核人员对每条告警标记“确认风险”或“误报”，这些结果要回流到指标体系里，用于评估每条规则的精确率，也能指导阈值调整。我用另一个轻量级 Flink 作业消费审核结果，按小时统计各规则的命中量、确认量、误报量，产出实时指标供风控平台展示。到这里，整条链路才真正跑通：业务事件进来，规则判定，告警降噪，人工复核，指标反馈，规则迭代。整个系统才活了。

我个人的体会是，欺诈检测作业最难的部分从来不是 Flink 语法本身，而是在于把业务里模糊的“可疑”翻译成状态、定时器和规则阈值，然后让作业在长期运行中保持稳定。DataStream API 的价值正在这里：它给了你足够的底层控制力来应对状态累积、乱序到达、动态规则这些真实问题。规则阈值这种看起来再简单不过的字段，在流作业里牵一发而动全身。如果这篇文章能让你少走几个我走过的弯路，那它就值得了。