1. 从一条广告点击日志说起这个统计任务到底在算什么广告点击量统计这件事听起来像是数据平台里最基础的一环但真正落到实时计算场景里坑远比想象中多。我最早接触这个需求的时候第一反应是“不就是按广告位分组计数吗”结果上线第一天就被脏数据教做人——某个广告位在十分钟内被同一个设备点了两万多次下游报表直接爆表。后来才明白点击量统计的核心难点从来不是计数本身而是如何在数据洪流里把无效点击、恶意点击、重复点击干净利落地剔除掉。这篇内容要聊的就是基于 Flink 做页面广告点击量统计并且把黑名单过滤这条链路完整地串起来。所谓黑名单过滤指的是在统计之前先把那些已知的恶意用户、异常设备、刷量 IP 从数据流里筛掉只让“干净”的点击进入计数逻辑。它解决的是数据质量问题适合已经对 Flink 有基本了解、正在做实时指标统计、并且被脏数据困扰的开发者参考。如果你还在用批处理跑 T1 的报表或者实时任务里完全没有过滤逻辑那这套思路值得你花时间看完。我先把整体链路用一句话说清楚广告点击事件从消息队列进来先经过黑名单过滤算子再按广告位维度做窗口聚合最后把统计结果写到下游存储。听起来简单但每一步都有讲究。比如黑名单数据从哪来、怎么动态更新、过滤算子放在哪个位置、窗口用滚动还是滑动、状态怎么管理、迟到数据怎么处理这些细节决定了你的任务能不能稳定跑下去。这里有个很多人一开始会忽略的点黑名单过滤和点击量统计本质上是一对矛盾体。过滤要求尽可能严格宁可错杀不可放过统计又要求尽可能完整不能漏掉真实点击。怎么平衡我的经验是分层处理——强黑名单直接丢弃弱黑名单打标但不丢弃让下游根据业务需要决定是否计入。这个思路后面会详细展开。另外要说明的是本文涉及的所有代码和配置都是基于常见实践整理的示例实际落地时需要根据你的数据规模、集群资源和业务规则做调整。我不会给你一套“复制就能跑”的万能模板因为实时计算这行脱离具体场景谈方案就是耍流氓。但我会把每个决策背后的逻辑讲透让你知道为什么这么选以及不这么选会踩什么坑。2. 黑名单过滤到底该放在哪一层位置选择的三种方案对比2.1 为什么过滤位置是个需要认真对待的问题很多人拿到需求第一反应是“在数据源后面加个 filter 算子不就完了”。这个想法没错但太粗糙。过滤位置的选择直接影响三件事状态大小、计算开销、以及黑名单更新的实时性。我见过有团队把黑名单过滤放在窗口聚合之后结果窗口里已经算进了脏数据再过滤只能把整个窗口结果丢掉等于白算。也见过把黑名单加载到每个算子里做本地缓存的黑名单一更新就得重启任务运维成本极高。所以过滤位置不是随便放的它取决于你的黑名单规模、更新频率、以及你对延迟的容忍度。下面我把常见的三种方案拆开讲每种都给出适用场景和实测表现。2.2 方案一数据源接入后立即过滤这是最直观的做法在 Source 之后直接接一个ProcessFunction或者RichFilterFunction每条数据进来先查黑名单命中就丢弃不命中就往下游发。DataStreamAdClickEvent filteredStream sourceStream .filter(new RichFilterFunctionAdClickEvent() { private transient ValueStateBoolean blacklistState; Override public void open(Configuration parameters) { // 初始化状态描述符用于存储黑名单标记 ValueStateDescriptorBoolean descriptor new ValueStateDescriptor(blacklist, Boolean.class); blacklistState getRuntimeContext().getState(descriptor); } Override public boolean filter(AdClickEvent event) throws Exception { // 查询外部黑名单或本地缓存 boolean isBlacklisted BlacklistCache.contains(event.getUserId()); return !isBlacklisted; } });这个方案的好处是脏数据最早被拦截下游所有算子都不用为无效数据付出计算成本。如果你的黑名单规模不大比如几万到几十万条完全可以把它加载到每个 TaskManager 的堆内缓存里查询走本地内存延迟在微秒级。但问题也很明显黑名单更新不及时。本地缓存意味着每个并行子任务都持有一份副本黑名单变更后需要广播到所有节点Flink 里可以用广播状态Broadcast State来实现但广播本身有延迟而且黑名单越大广播开销越高。我实测过一个百万级黑名单用广播状态更新每次全量广播要几秒钟期间新数据可能漏过过滤。提示如果黑名单更新频率低于每分钟一次且规模在十万级以内这个方案完全够用。但如果黑名单是实时变化的比如风控系统每秒都在产出新的恶意用户就得考虑下面的方案。2.3 方案二旁路查询外部存储把黑名单放在外部存储比如 Redis、HBase过滤算子每条数据都去查一次。这样黑名单更新是实时的风控系统写入后立刻生效。public class BlacklistQueryFunction extends RichFlatMapFunctionAdClickEvent, AdClickEvent { private transient Jedis jedis; Override public void open(Configuration parameters) { // 初始化连接池注意每个并行子任务独立连接 jedis new Jedis(blacklist-host, 6379); } Override public void flatMap(AdClickEvent event, CollectorAdClickEvent out) { // 用 userId 作为 key 查询黑名单 Boolean exists jedis.sismember(ad:blacklist:users, event.getUserId()); if (exists null || !exists) { out.collect(event); } // 命中黑名单的直接丢弃不往下游发 } }这个方案的优点是黑名单实时性最好且不受本地内存限制理论上可以支撑千万级甚至更大的黑名单。但代价是每条数据都要走一次网络 IO如果 QPS 很高Redis 很容易成为瓶颈。我压测过一个场景单 Redis 实例在 5 万 QPS 下响应时间开始明显抖动超过 8 万 QPS 就出现超时。优化思路有两个一是加本地缓存做多级查询热点黑名单放本地冷门数据走 Redis二是用异步 IOFlink 的AsyncDataStream来避免同步等待阻塞算子线程。异步 IO 的写法这里不展开核心是把查询请求批量发出去用回调处理结果能显著提升吞吐。2.4 方案三广播状态动态更新这是 Flink 官方推荐的动态规则匹配方案。把黑名单作为广播流主数据流和广播流连接connect在KeyedBroadcastProcessFunction里用广播状态存储黑名单每条数据进来查状态即可。// 黑名单广播流 BroadcastStreamString blacklistBroadcast blacklistStream .broadcast(blacklistStateDescriptor); // 主数据流连接广播流 DataStreamAdClickEvent filtered clickStream .connect(blacklistBroadcast) .process(new BlacklistFilterFunction()); // 过滤函数核心逻辑 public class BlacklistFilterFunction extends KeyedBroadcastProcessFunctionString, AdClickEvent, String, AdClickEvent { Override public void processElement(AdClickEvent event, ReadOnlyContext ctx, CollectorAdClickEvent out) throws Exception { // 从广播状态读取黑名单 ReadOnlyBroadcastStateString, Boolean state ctx.getBroadcastState(blacklistStateDescriptor); if (state.get(event.getUserId()) null) { out.collect(event); } } Override public void processBroadcastElement(String userId, Context ctx, CollectorAdClickEvent out) throws Exception { // 更新广播状态 BroadcastStateString, Boolean state ctx.getBroadcastState(blacklistStateDescriptor); state.put(userId, true); } }这个方案兼顾了实时性和性能黑名单更新通过广播流实时同步查询走本地状态没有网络开销。但要注意广播状态是存在每个并行子任务的内存里的黑名单太大会撑爆内存。我的经验是广播状态适合百万级以内的黑名单再大就得考虑分片或者换方案。三种方案的对比我整理成表格方便你按场景选方案黑名单规模更新实时性查询延迟运维复杂度适用场景本地缓存过滤十万级以内分钟级微秒级低黑名单稳定、QPS 高旁路查询外部存储千万级秒级毫秒级中黑名单频繁更新广播状态百万级以内秒级微秒级中高实时性与性能兼顾2.5 我的实际选择与理由在最近一个广告点击统计项目里我最终用的是广播状态 本地缓存兜底的混合方案。具体来说风控系统产出的黑名单通过广播流实时同步到过滤算子同时算子内部维护一个 LRU 本地缓存缓存最近命中的黑名单用户。这样即使广播有短暂延迟本地缓存也能挡住大部分重复的恶意点击。为什么不选旁路查询因为那个项目的点击 QPS 峰值到了 12 万Redis 扛不住。为什么不选纯本地缓存因为风控要求黑名单变更后 5 秒内生效本地缓存的广播延迟满足不了。混合方案是在性能和实时性之间找到的平衡点实测下来过滤延迟稳定在 2 秒以内CPU 开销比纯旁路查询低了 60%。3. 点击量统计的窗口设计滚动、滑动还是会话窗口3.1 窗口类型的选择逻辑过滤完之后干净的数据要进入窗口做聚合。Flink 提供了三种常用窗口滚动窗口Tumbling、滑动窗口Sliding、会话窗口Session。选哪种取决于你的报表需求。滚动窗口是固定长度、不重叠的窗口比如每 5 分钟统计一次点击量。它的优点是实现简单、状态可控每个事件只属于一个窗口。缺点是统计结果是“跳变”的5 分钟整点才输出一次中间没有更新。滑动窗口是固定长度、有重叠的窗口比如窗口长度 10 分钟、滑动步长 1 分钟。它每 1 分钟输出一次过去 10 分钟的统计结果曲线更平滑。但代价是每个事件会属于多个窗口状态和计算量成倍增加。我实测过滑动步长是窗口长度的 1/10 时状态大小大约是滚动窗口的 10 倍。会话窗口是按活动间隙切分的比如用户停止点击 30 秒后认为会话结束。它适合分析用户行为序列但不太适合广告点击量这种固定周期的统计需求。我的建议很直接广告点击量统计优先用滚动窗口。因为报表通常就是按固定周期出数的滚动窗口的状态管理最简单出问题时排查也容易。如果业务方要求“实时曲线”可以用滑动窗口但一定要控制好步长和窗口长度的比例别搞出 1 秒步长、1 小时窗口这种配置状态会爆炸。3.2 窗口聚合的代码实现与增量计算滚动窗口的聚合代码大概长这样DataStreamAdClickCount windowedCount filteredStream .keyBy(AdClickEvent::getAdSlotId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .aggregate(new ClickCountAggregator(), new ClickCountWindowFunction());这里有个关键点用aggregate而不是apply。aggregate是增量聚合每来一条数据就更新累加器窗口触发时只输出最终结果状态里只存累加器一个 long 值。apply是全量聚合窗口触发时把所有数据拉出来遍历状态里存的是所有原始事件。两者在状态大小和性能上差距巨大数据量大时apply基本不可用。// 增量聚合函数每来一条数据就累加 public class ClickCountAggregator implements AggregateFunctionAdClickEvent, Long, Long { Override public Long createAccumulator() { return 0L; } Override public Long add(AdClickEvent event, Long accumulator) { return accumulator 1; } Override public Long getResult(Long accumulator) { return accumulator; } Override public Long merge(Long a, Long b) { return a b; } }这个累加器只占 8 字节即使有几十万个广告位状态也就几 MB。如果用apply每个窗口要缓存所有事件内存占用是数量级的差距。3.3 事件时间与水位线的处理广告点击统计必须用事件时间不能用处理时间。因为数据从产生到进入 Flink 有延迟用处理时间会导致统计结果和实际发生时间对不上。事件时间需要配合水位线Watermark使用水位线告诉 Flink “某个时间点之前的数据都到齐了”。DataStreamAdClickEvent withWatermark stream .assignTimestampsAndWatermarks( WatermarkStrategy.AdClickEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) - event.getClickTime()) );forBoundedOutOfOrderness表示允许数据最多迟到 5 秒。这个值怎么定我的经验是看数据源的乱序程度。如果消息队列本身有序1 秒就够如果经过多个中间环节可能需要 10 秒甚至更多。设太小会丢迟到数据设太大会增加窗口触发延迟。注意水位线是全局推进的如果某个并行子任务没有数据它的水位线不会前进会拖慢整个任务。解决办法是设置withIdleness让空闲子任务的水位线自动推进。3.4 迟到数据的三种处理策略即使设了水位线还是会有迟到数据。Flink 提供了三种处理方式默认丢弃窗口触发后到达的数据直接扔掉。适合对完整性要求不高的场景。allowedLateness允许窗口触发后再保留一段时间迟到数据还能触发更新。比如设 1 分钟窗口触发后 1 分钟内的迟到数据会重新计算并输出更新结果。sideOutput把迟到数据输出到侧流单独处理。适合需要记录迟到数据做分析的场景。OutputTagAdClickEvent lateTag new OutputTagAdClickEvent(late-data){}; SingleOutputStreamOperatorAdClickCount result filteredStream .keyBy(AdClickEvent::getAdSlotId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .allowedLateness(Time.minutes(1)) .sideOutputLateData(lateTag) .aggregate(new ClickCountAggregator(), new ClickCountWindowFunction()); DataStreamAdClickEvent lateStream result.getSideOutput(lateTag);我一般会同时开allowedLateness和sideOutput前者保证统计结果尽量准确后者用来监控迟到数据的量。如果侧流数据量突然增大说明数据源出了问题需要排查。4. 状态管理与容错让任务能稳定跑上几个月4.1 状态后端的选择Flink 的状态后端决定了状态存在哪、怎么存。常见的有三种HashMapStateBackend内存、EmbeddedRocksDBStateBackendRocksDB、以及基于文件系统的状态后端。对于广告点击统计这种场景状态大小通常在几百 MB 到几 GB 之间。如果状态能放进内存用HashMapStateBackend性能最好读写都是内存操作。但如果状态可能超过内存就得用RocksDBStateBackend它把状态存在本地磁盘支持超出内存的状态量代价是读写有磁盘 IO 开销。// 配置 RocksDB 状态后端 Configuration config new Configuration(); config.setString(state.backend, rocksdb); config.setString(state.backend.rocksdb.localdir, /data/flink/rocksdb); config.setString(state.checkpoints.dir, hdfs:///flink/checkpoints);我的经验是如果广告位数量在十万级以内用内存状态后端超过十万或者窗口状态很大用 RocksDB。另外 RocksDB 的本地目录一定要配到 SSD 上机械硬盘的随机读写会让性能下降一个数量级。4.2 Checkpoint 配置与调优Checkpoint 是 Flink 容错的核心机制它定期把状态快照写到持久化存储。配置不当会导致 Checkpoint 超时、失败甚至影响正常数据处理。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(60000); // 每 60 秒一次 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); // 两次间隔至少 30 秒 env.getCheckpointConfig().setCheckpointTimeout(120000); // 超时 2 分钟 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 不并发 env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); // 容忍 3 次失败几个关键参数的解释Checkpoint 间隔太短会增加开销太长会导致故障恢复时重放数据过多。我一般设 1 到 5 分钟看数据量和恢复时间要求。最小间隔防止前一个 Checkpoint 还没完后一个就开始了。设成间隔的一半左右比较稳妥。超时时间如果状态很大写 HDFS 可能超过默认的 10 分钟需要调大。容忍失败次数偶尔失败不用让整个任务挂掉设个容忍值更稳。提示Checkpoint 失败最常见的原因是状态太大导致写入超时。如果频繁失败先看状态大小再考虑调大超时或优化状态结构。4.3 重启策略与故障恢复任务挂了怎么恢复Flink 提供了多种重启策略// 固定延迟重启最多重启 5 次每次间隔 10 秒 env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 5, Time.seconds(10) )); // 指数延迟重启间隔逐渐增大适合避免频繁重启打爆集群 env.setRestartStrategy(RestartStrategies.exponentialDelayRestart( Time.seconds(1), Time.minutes(5), 2.0, Time.hours(1), 10 ));我一般用指数延迟重启因为如果任务是因为代码 bug 挂的固定间隔重启会反复失败指数延迟能给运维争取排查时间。另外要配合告警任务重启次数超过阈值就通知人介入。4.4 大状态下的性能优化状态大了之后性能问题会逐渐暴露。我踩过的几个坑第一个坑是状态序列化开销。Flink 默认用 Java 序列化效率低。换成 Kryo 或者自定义序列化器能提升不少。对于简单的 POJO实现Serializable接口并注册到 Kryo 就行。第二个坑是 RocksDB 的写放大。RocksDB 的 LSM 树结构在写入量大时会有明显的写放大导致磁盘 IO 飙升。解决办法是调大write_buffer_size和max_write_buffer_number让更多数据在内存里合并后再落盘。第三个坑是 Checkpoint 期间的背压。Checkpoint 时状态要快照如果状态太大快照期间数据处理会变慢产生背压。可以开启增量 CheckpointRocksDB 支持只上传变化的部分大幅减少快照数据量。// 开启 RocksDB 增量 Checkpoint env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION ); config.setString(state.backend.rocksdb.incremental, true);5. 黑名单过滤与统计链路的联调实测5.1 测试数据的构造联调阶段最麻烦的是造数据。真实广告点击数据不好拿我一般用脚本生成模拟数据关键是要覆盖各种边界情况正常点击随机用户、随机广告位、时间戳递增黑名单用户点击用已知的黑名单 ID 构造重复点击同一用户对同一广告位短时间多次点击迟到数据时间戳明显早于当前水位线的数据乱序数据时间戳不按顺序到达import random import time import json def generate_click_event(): user_id fuser_{random.randint(1, 100000)} ad_slot_id fslot_{random.randint(1, 1000)} click_time int(time.time() * 1000) - random.randint(0, 10000) return json.dumps({ userId: user_id, adSlotId: ad_slot_id, clickTime: click_time }) # 生成 10 万条测试数据 with open(test_clicks.json, w) as f: for _ in range(100000): f.write(generate_click_event() \n)5.2 过滤效果的验证方法怎么确认黑名单过滤真的生效了我的做法是在过滤算子前后各埋一个计数器统计进入和流出的数据量差值就是被过滤掉的数量。public class BlacklistFilterFunction extends RichFilterFunctionAdClickEvent { private transient LongCounter inputCounter; private transient LongCounter filteredCounter; Override public void open(Configuration parameters) { inputCounter getRuntimeContext().getLongCounter(input_count); filteredCounter getRuntimeContext().getLongCounter(filtered_count); } Override public boolean filter(AdClickEvent event) throws Exception { inputCounter.add(1); boolean isBlacklisted BlacklistCache.contains(event.getUserId()); if (isBlacklisted) { filteredCounter.add(1); } return !isBlacklisted; } }然后在 Flink Web UI 的 Metrics 页面看这两个计数器的值。如果filtered_count一直是 0要么是黑名单没加载成功要么是测试数据里没有黑名单用户。我遇到过黑名单广播流因为序列化问题一直没数据的情况就是靠这个计数器发现的。5.3 统计结果的核对窗口输出的统计结果要和预期值核对。最简单的办法是用批处理跑一遍同样的数据对比结果。如果实时和批处理的差值在允许范围内考虑迟到数据说明统计逻辑没问题。另一个办法是抽样核对从原始数据里随机抽几个广告位手动数一下点击量和 Flink 输出的结果对比。这个方法适合数据量不大的场景能发现一些聚合逻辑的细节问题。我踩过的一个坑是窗口边界问题滚动窗口是左闭右开的比如[00:00, 00:05)00:05:00 整点的数据会落到下一个窗口。如果业务方理解成闭区间对账时就会差一条。这种问题一定要在联调阶段和业务方确认清楚。5.4 压力测试与性能瓶颈定位上线前必须做压力测试。我用的是自己写的压测脚本往消息队列里灌数据逐步加大 QPS观察 Flink 任务的各项指标指标正常范围异常表现可能原因吞吐量随 QPS 线性增长增长停滞算子成为瓶颈背压0 或低持续高背压下游处理慢Checkpoint 时长稳定逐渐变长状态膨胀CPU 使用率60% 以下持续 90%计算密集或 GCGC 时间低频繁 Full GC内存不足或泄漏背压是最直观的指标。Flink Web UI 里每个算子都有背压指示红色表示背压严重。如果过滤算子背压高说明黑名单查询慢如果窗口算子背压高说明聚合逻辑重或者状态大。我遇到过一次窗口算子背压排查后发现是allowedLateness设得太长10 分钟导致窗口状态一直不释放。改成 1 分钟后背压立刻消失。这个参数不是越大越好要根据数据迟到情况合理设置。6. 几个容易翻车的细节和我的应对经验6.1 黑名单数据的格式一致性黑名单从风控系统同步过来格式一定要统一。我遇到过风控系统有时候发 JSON、有时候发纯文本的情况导致广播流解析失败黑名单更新中断。解决办法是在广播流入口加一层格式校验解析失败的数据写到侧流告警不让它影响主流程。DataStreamString validatedBlacklist rawBlacklist .process(new ProcessFunctionString, String() { Override public void processElement(String value, Context ctx, CollectorString out) { try { // 尝试解析验证格式 JSON.parseObject(value); out.collect(value); } catch (Exception e) { // 格式错误输出到侧流 ctx.output(new OutputTagString(invalid-blacklist){}, value); } } });6.2 黑名单删除的同步问题黑名单不只有新增还有删除比如误判后解封。如果广播流只发新增删除操作就同步不过去。我的做法是广播流里带上操作类型ADD表示加入黑名单REMOVE表示移出。Override public void processBroadcastElement(BlacklistOp op, Context ctx, CollectorAdClickEvent out) throws Exception { BroadcastStateString, Boolean state ctx.getBroadcastState(descriptor); if (ADD.equals(op.getType())) { state.put(op.getUserId(), true); } else if (REMOVE.equals(op.getType())) { state.remove(op.getUserId()); } }这个细节很容易被忽略上线后才发现解封的用户还是被过滤业务方投诉才回头改。6.3 窗口结果输出的幂等性统计结果写到下游存储时要考虑幂等性。因为 Checkpoint 恢复后窗口可能重新触发同一个窗口的结果会输出两次。如果下游是数据库用INSERT ... ON DUPLICATE KEY UPDATE或者先删后插来保证幂等。如果下游是消息队列消费端要做去重。我一般会在结果里带上窗口的起止时间下游按广告位 窗口起始时间做主键重复写入直接覆盖。6.4 监控告警的配置任务上线不是终点监控才是。我必配的几个告警Checkpoint 连续失败超过 3 次就告警可能是状态太大或存储有问题。背压持续过高某个算子背压超过 5 分钟说明处理能力不足。黑名单过滤量为 0如果一段时间内过滤量突然归零可能是黑名单同步断了。任务重启非预期重启要立即通知排查是资源问题还是代码 bug。监控指标可以从 Flink 的 Metrics 系统导出到 Prometheus再用 Grafana 做面板。这套组合用起来很顺手社区里资料也多。6.5 资源规划的保守原则最后说资源规划。我见过太多任务因为资源给少了上线后频繁背压、Checkpoint 失败。我的原则是按峰值 QPS 的 1.5 倍规划资源并且留出至少 30% 的内存余量给状态和 GC。具体来说如果峰值 QPS 是 10 万单并行度能处理 1 万 QPS那至少开 15 个并行度。内存方面如果状态预估 2 GBTaskManager 内存至少给 4 GB留一半给堆外和 GC。这个配置不是拍脑袋是压测出来的经验值。资源给多了浪费给少了出事宁可稍微宽裕一点。广告点击量统计这个任务技术栈不复杂但细节特别多。黑名单过滤、窗口聚合、状态管理、容错恢复每一环都有坑。我把这几年的经验整理出来希望能帮你少走点弯路。实际落地时建议先在测试环境把链路跑通用模拟数据验证过滤和统计逻辑再做压力测试确定资源配置最后小流量上线观察一段时间。稳扎稳打比一步到位靠谱得多。
1. 从一条广告点击日志说起这个统计任务到底在算什么广告点击量统计这件事听起来像是数据平台里最基础的一环但真正落到实时计算场景里坑远比想象中多。我最早接触这个需求的时候第一反应是“不就是按广告位分组计数吗”结果上线第一天就被脏数据教做人——某个广告位在十分钟内被同一个设备点了两万多次下游报表直接爆表。后来才明白点击量统计的核心难点从来不是计数本身而是如何在数据洪流里把无效点击、恶意点击、重复点击干净利落地剔除掉。这篇内容要聊的就是基于 Flink 做页面广告点击量统计并且把黑名单过滤这条链路完整地串起来。所谓黑名单过滤指的是在统计之前先把那些已知的恶意用户、异常设备、刷量 IP 从数据流里筛掉只让“干净”的点击进入计数逻辑。它解决的是数据质量问题适合已经对 Flink 有基本了解、正在做实时指标统计、并且被脏数据困扰的开发者参考。如果你还在用批处理跑 T1 的报表或者实时任务里完全没有过滤逻辑那这套思路值得你花时间看完。我先把整体链路用一句话说清楚广告点击事件从消息队列进来先经过黑名单过滤算子再按广告位维度做窗口聚合最后把统计结果写到下游存储。听起来简单但每一步都有讲究。比如黑名单数据从哪来、怎么动态更新、过滤算子放在哪个位置、窗口用滚动还是滑动、状态怎么管理、迟到数据怎么处理这些细节决定了你的任务能不能稳定跑下去。这里有个很多人一开始会忽略的点黑名单过滤和点击量统计本质上是一对矛盾体。过滤要求尽可能严格宁可错杀不可放过统计又要求尽可能完整不能漏掉真实点击。怎么平衡我的经验是分层处理——强黑名单直接丢弃弱黑名单打标但不丢弃让下游根据业务需要决定是否计入。这个思路后面会详细展开。另外要说明的是本文涉及的所有代码和配置都是基于常见实践整理的示例实际落地时需要根据你的数据规模、集群资源和业务规则做调整。我不会给你一套“复制就能跑”的万能模板因为实时计算这行脱离具体场景谈方案就是耍流氓。但我会把每个决策背后的逻辑讲透让你知道为什么这么选以及不这么选会踩什么坑。2. 黑名单过滤到底该放在哪一层位置选择的三种方案对比2.1 为什么过滤位置是个需要认真对待的问题很多人拿到需求第一反应是“在数据源后面加个 filter 算子不就完了”。这个想法没错但太粗糙。过滤位置的选择直接影响三件事状态大小、计算开销、以及黑名单更新的实时性。我见过有团队把黑名单过滤放在窗口聚合之后结果窗口里已经算进了脏数据再过滤只能把整个窗口结果丢掉等于白算。也见过把黑名单加载到每个算子里做本地缓存的黑名单一更新就得重启任务运维成本极高。所以过滤位置不是随便放的它取决于你的黑名单规模、更新频率、以及你对延迟的容忍度。下面我把常见的三种方案拆开讲每种都给出适用场景和实测表现。2.2 方案一数据源接入后立即过滤这是最直观的做法在 Source 之后直接接一个ProcessFunction或者RichFilterFunction每条数据进来先查黑名单命中就丢弃不命中就往下游发。DataStreamAdClickEvent filteredStream sourceStream .filter(new RichFilterFunctionAdClickEvent() { private transient ValueStateBoolean blacklistState; Override public void open(Configuration parameters) { // 初始化状态描述符用于存储黑名单标记 ValueStateDescriptorBoolean descriptor new ValueStateDescriptor(blacklist, Boolean.class); blacklistState getRuntimeContext().getState(descriptor); } Override public boolean filter(AdClickEvent event) throws Exception { // 查询外部黑名单或本地缓存 boolean isBlacklisted BlacklistCache.contains(event.getUserId()); return !isBlacklisted; } });这个方案的好处是脏数据最早被拦截下游所有算子都不用为无效数据付出计算成本。如果你的黑名单规模不大比如几万到几十万条完全可以把它加载到每个 TaskManager 的堆内缓存里查询走本地内存延迟在微秒级。但问题也很明显黑名单更新不及时。本地缓存意味着每个并行子任务都持有一份副本黑名单变更后需要广播到所有节点Flink 里可以用广播状态Broadcast State来实现但广播本身有延迟而且黑名单越大广播开销越高。我实测过一个百万级黑名单用广播状态更新每次全量广播要几秒钟期间新数据可能漏过过滤。提示如果黑名单更新频率低于每分钟一次且规模在十万级以内这个方案完全够用。但如果黑名单是实时变化的比如风控系统每秒都在产出新的恶意用户就得考虑下面的方案。2.3 方案二旁路查询外部存储把黑名单放在外部存储比如 Redis、HBase过滤算子每条数据都去查一次。这样黑名单更新是实时的风控系统写入后立刻生效。public class BlacklistQueryFunction extends RichFlatMapFunctionAdClickEvent, AdClickEvent { private transient Jedis jedis; Override public void open(Configuration parameters) { // 初始化连接池注意每个并行子任务独立连接 jedis new Jedis(blacklist-host, 6379); } Override public void flatMap(AdClickEvent event, CollectorAdClickEvent out) { // 用 userId 作为 key 查询黑名单 Boolean exists jedis.sismember(ad:blacklist:users, event.getUserId()); if (exists null || !exists) { out.collect(event); } // 命中黑名单的直接丢弃不往下游发 } }这个方案的优点是黑名单实时性最好且不受本地内存限制理论上可以支撑千万级甚至更大的黑名单。但代价是每条数据都要走一次网络 IO如果 QPS 很高Redis 很容易成为瓶颈。我压测过一个场景单 Redis 实例在 5 万 QPS 下响应时间开始明显抖动超过 8 万 QPS 就出现超时。优化思路有两个一是加本地缓存做多级查询热点黑名单放本地冷门数据走 Redis二是用异步 IOFlink 的AsyncDataStream来避免同步等待阻塞算子线程。异步 IO 的写法这里不展开核心是把查询请求批量发出去用回调处理结果能显著提升吞吐。2.4 方案三广播状态动态更新这是 Flink 官方推荐的动态规则匹配方案。把黑名单作为广播流主数据流和广播流连接connect在KeyedBroadcastProcessFunction里用广播状态存储黑名单每条数据进来查状态即可。// 黑名单广播流 BroadcastStreamString blacklistBroadcast blacklistStream .broadcast(blacklistStateDescriptor); // 主数据流连接广播流 DataStreamAdClickEvent filtered clickStream .connect(blacklistBroadcast) .process(new BlacklistFilterFunction()); // 过滤函数核心逻辑 public class BlacklistFilterFunction extends KeyedBroadcastProcessFunctionString, AdClickEvent, String, AdClickEvent { Override public void processElement(AdClickEvent event, ReadOnlyContext ctx, CollectorAdClickEvent out) throws Exception { // 从广播状态读取黑名单 ReadOnlyBroadcastStateString, Boolean state ctx.getBroadcastState(blacklistStateDescriptor); if (state.get(event.getUserId()) null) { out.collect(event); } } Override public void processBroadcastElement(String userId, Context ctx, CollectorAdClickEvent out) throws Exception { // 更新广播状态 BroadcastStateString, Boolean state ctx.getBroadcastState(blacklistStateDescriptor); state.put(userId, true); } }这个方案兼顾了实时性和性能黑名单更新通过广播流实时同步查询走本地状态没有网络开销。但要注意广播状态是存在每个并行子任务的内存里的黑名单太大会撑爆内存。我的经验是广播状态适合百万级以内的黑名单再大就得考虑分片或者换方案。三种方案的对比我整理成表格方便你按场景选方案黑名单规模更新实时性查询延迟运维复杂度适用场景本地缓存过滤十万级以内分钟级微秒级低黑名单稳定、QPS 高旁路查询外部存储千万级秒级毫秒级中黑名单频繁更新广播状态百万级以内秒级微秒级中高实时性与性能兼顾2.5 我的实际选择与理由在最近一个广告点击统计项目里我最终用的是广播状态 本地缓存兜底的混合方案。具体来说风控系统产出的黑名单通过广播流实时同步到过滤算子同时算子内部维护一个 LRU 本地缓存缓存最近命中的黑名单用户。这样即使广播有短暂延迟本地缓存也能挡住大部分重复的恶意点击。为什么不选旁路查询因为那个项目的点击 QPS 峰值到了 12 万Redis 扛不住。为什么不选纯本地缓存因为风控要求黑名单变更后 5 秒内生效本地缓存的广播延迟满足不了。混合方案是在性能和实时性之间找到的平衡点实测下来过滤延迟稳定在 2 秒以内CPU 开销比纯旁路查询低了 60%。3. 点击量统计的窗口设计滚动、滑动还是会话窗口3.1 窗口类型的选择逻辑过滤完之后干净的数据要进入窗口做聚合。Flink 提供了三种常用窗口滚动窗口Tumbling、滑动窗口Sliding、会话窗口Session。选哪种取决于你的报表需求。滚动窗口是固定长度、不重叠的窗口比如每 5 分钟统计一次点击量。它的优点是实现简单、状态可控每个事件只属于一个窗口。缺点是统计结果是“跳变”的5 分钟整点才输出一次中间没有更新。滑动窗口是固定长度、有重叠的窗口比如窗口长度 10 分钟、滑动步长 1 分钟。它每 1 分钟输出一次过去 10 分钟的统计结果曲线更平滑。但代价是每个事件会属于多个窗口状态和计算量成倍增加。我实测过滑动步长是窗口长度的 1/10 时状态大小大约是滚动窗口的 10 倍。会话窗口是按活动间隙切分的比如用户停止点击 30 秒后认为会话结束。它适合分析用户行为序列但不太适合广告点击量这种固定周期的统计需求。我的建议很直接广告点击量统计优先用滚动窗口。因为报表通常就是按固定周期出数的滚动窗口的状态管理最简单出问题时排查也容易。如果业务方要求“实时曲线”可以用滑动窗口但一定要控制好步长和窗口长度的比例别搞出 1 秒步长、1 小时窗口这种配置状态会爆炸。3.2 窗口聚合的代码实现与增量计算滚动窗口的聚合代码大概长这样DataStreamAdClickCount windowedCount filteredStream .keyBy(AdClickEvent::getAdSlotId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .aggregate(new ClickCountAggregator(), new ClickCountWindowFunction());这里有个关键点用aggregate而不是apply。aggregate是增量聚合每来一条数据就更新累加器窗口触发时只输出最终结果状态里只存累加器一个 long 值。apply是全量聚合窗口触发时把所有数据拉出来遍历状态里存的是所有原始事件。两者在状态大小和性能上差距巨大数据量大时apply基本不可用。// 增量聚合函数每来一条数据就累加 public class ClickCountAggregator implements AggregateFunctionAdClickEvent, Long, Long { Override public Long createAccumulator() { return 0L; } Override public Long add(AdClickEvent event, Long accumulator) { return accumulator 1; } Override public Long getResult(Long accumulator) { return accumulator; } Override public Long merge(Long a, Long b) { return a b; } }这个累加器只占 8 字节即使有几十万个广告位状态也就几 MB。如果用apply每个窗口要缓存所有事件内存占用是数量级的差距。3.3 事件时间与水位线的处理广告点击统计必须用事件时间不能用处理时间。因为数据从产生到进入 Flink 有延迟用处理时间会导致统计结果和实际发生时间对不上。事件时间需要配合水位线Watermark使用水位线告诉 Flink “某个时间点之前的数据都到齐了”。DataStreamAdClickEvent withWatermark stream .assignTimestampsAndWatermarks( WatermarkStrategy.AdClickEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) - event.getClickTime()) );forBoundedOutOfOrderness表示允许数据最多迟到 5 秒。这个值怎么定我的经验是看数据源的乱序程度。如果消息队列本身有序1 秒就够如果经过多个中间环节可能需要 10 秒甚至更多。设太小会丢迟到数据设太大会增加窗口触发延迟。注意水位线是全局推进的如果某个并行子任务没有数据它的水位线不会前进会拖慢整个任务。解决办法是设置withIdleness让空闲子任务的水位线自动推进。3.4 迟到数据的三种处理策略即使设了水位线还是会有迟到数据。Flink 提供了三种处理方式默认丢弃窗口触发后到达的数据直接扔掉。适合对完整性要求不高的场景。allowedLateness允许窗口触发后再保留一段时间迟到数据还能触发更新。比如设 1 分钟窗口触发后 1 分钟内的迟到数据会重新计算并输出更新结果。sideOutput把迟到数据输出到侧流单独处理。适合需要记录迟到数据做分析的场景。OutputTagAdClickEvent lateTag new OutputTagAdClickEvent(late-data){}; SingleOutputStreamOperatorAdClickCount result filteredStream .keyBy(AdClickEvent::getAdSlotId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .allowedLateness(Time.minutes(1)) .sideOutputLateData(lateTag) .aggregate(new ClickCountAggregator(), new ClickCountWindowFunction()); DataStreamAdClickEvent lateStream result.getSideOutput(lateTag);我一般会同时开allowedLateness和sideOutput前者保证统计结果尽量准确后者用来监控迟到数据的量。如果侧流数据量突然增大说明数据源出了问题需要排查。4. 状态管理与容错让任务能稳定跑上几个月4.1 状态后端的选择Flink 的状态后端决定了状态存在哪、怎么存。常见的有三种HashMapStateBackend内存、EmbeddedRocksDBStateBackendRocksDB、以及基于文件系统的状态后端。对于广告点击统计这种场景状态大小通常在几百 MB 到几 GB 之间。如果状态能放进内存用HashMapStateBackend性能最好读写都是内存操作。但如果状态可能超过内存就得用RocksDBStateBackend它把状态存在本地磁盘支持超出内存的状态量代价是读写有磁盘 IO 开销。// 配置 RocksDB 状态后端 Configuration config new Configuration(); config.setString(state.backend, rocksdb); config.setString(state.backend.rocksdb.localdir, /data/flink/rocksdb); config.setString(state.checkpoints.dir, hdfs:///flink/checkpoints);我的经验是如果广告位数量在十万级以内用内存状态后端超过十万或者窗口状态很大用 RocksDB。另外 RocksDB 的本地目录一定要配到 SSD 上机械硬盘的随机读写会让性能下降一个数量级。4.2 Checkpoint 配置与调优Checkpoint 是 Flink 容错的核心机制它定期把状态快照写到持久化存储。配置不当会导致 Checkpoint 超时、失败甚至影响正常数据处理。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(60000); // 每 60 秒一次 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); // 两次间隔至少 30 秒 env.getCheckpointConfig().setCheckpointTimeout(120000); // 超时 2 分钟 env.getCheckpointConfig().setMaxConcurrentCheckpoints(1); // 不并发 env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3); // 容忍 3 次失败几个关键参数的解释Checkpoint 间隔太短会增加开销太长会导致故障恢复时重放数据过多。我一般设 1 到 5 分钟看数据量和恢复时间要求。最小间隔防止前一个 Checkpoint 还没完后一个就开始了。设成间隔的一半左右比较稳妥。超时时间如果状态很大写 HDFS 可能超过默认的 10 分钟需要调大。容忍失败次数偶尔失败不用让整个任务挂掉设个容忍值更稳。提示Checkpoint 失败最常见的原因是状态太大导致写入超时。如果频繁失败先看状态大小再考虑调大超时或优化状态结构。4.3 重启策略与故障恢复任务挂了怎么恢复Flink 提供了多种重启策略// 固定延迟重启最多重启 5 次每次间隔 10 秒 env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 5, Time.seconds(10) )); // 指数延迟重启间隔逐渐增大适合避免频繁重启打爆集群 env.setRestartStrategy(RestartStrategies.exponentialDelayRestart( Time.seconds(1), Time.minutes(5), 2.0, Time.hours(1), 10 ));我一般用指数延迟重启因为如果任务是因为代码 bug 挂的固定间隔重启会反复失败指数延迟能给运维争取排查时间。另外要配合告警任务重启次数超过阈值就通知人介入。4.4 大状态下的性能优化状态大了之后性能问题会逐渐暴露。我踩过的几个坑第一个坑是状态序列化开销。Flink 默认用 Java 序列化效率低。换成 Kryo 或者自定义序列化器能提升不少。对于简单的 POJO实现Serializable接口并注册到 Kryo 就行。第二个坑是 RocksDB 的写放大。RocksDB 的 LSM 树结构在写入量大时会有明显的写放大导致磁盘 IO 飙升。解决办法是调大write_buffer_size和max_write_buffer_number让更多数据在内存里合并后再落盘。第三个坑是 Checkpoint 期间的背压。Checkpoint 时状态要快照如果状态太大快照期间数据处理会变慢产生背压。可以开启增量 CheckpointRocksDB 支持只上传变化的部分大幅减少快照数据量。// 开启 RocksDB 增量 Checkpoint env.getCheckpointConfig().enableExternalizedCheckpoints( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION ); config.setString(state.backend.rocksdb.incremental, true);5. 黑名单过滤与统计链路的联调实测5.1 测试数据的构造联调阶段最麻烦的是造数据。真实广告点击数据不好拿我一般用脚本生成模拟数据关键是要覆盖各种边界情况正常点击随机用户、随机广告位、时间戳递增黑名单用户点击用已知的黑名单 ID 构造重复点击同一用户对同一广告位短时间多次点击迟到数据时间戳明显早于当前水位线的数据乱序数据时间戳不按顺序到达import random import time import json def generate_click_event(): user_id fuser_{random.randint(1, 100000)} ad_slot_id fslot_{random.randint(1, 1000)} click_time int(time.time() * 1000) - random.randint(0, 10000) return json.dumps({ userId: user_id, adSlotId: ad_slot_id, clickTime: click_time }) # 生成 10 万条测试数据 with open(test_clicks.json, w) as f: for _ in range(100000): f.write(generate_click_event() \n)5.2 过滤效果的验证方法怎么确认黑名单过滤真的生效了我的做法是在过滤算子前后各埋一个计数器统计进入和流出的数据量差值就是被过滤掉的数量。public class BlacklistFilterFunction extends RichFilterFunctionAdClickEvent { private transient LongCounter inputCounter; private transient LongCounter filteredCounter; Override public void open(Configuration parameters) { inputCounter getRuntimeContext().getLongCounter(input_count); filteredCounter getRuntimeContext().getLongCounter(filtered_count); } Override public boolean filter(AdClickEvent event) throws Exception { inputCounter.add(1); boolean isBlacklisted BlacklistCache.contains(event.getUserId()); if (isBlacklisted) { filteredCounter.add(1); } return !isBlacklisted; } }然后在 Flink Web UI 的 Metrics 页面看这两个计数器的值。如果filtered_count一直是 0要么是黑名单没加载成功要么是测试数据里没有黑名单用户。我遇到过黑名单广播流因为序列化问题一直没数据的情况就是靠这个计数器发现的。5.3 统计结果的核对窗口输出的统计结果要和预期值核对。最简单的办法是用批处理跑一遍同样的数据对比结果。如果实时和批处理的差值在允许范围内考虑迟到数据说明统计逻辑没问题。另一个办法是抽样核对从原始数据里随机抽几个广告位手动数一下点击量和 Flink 输出的结果对比。这个方法适合数据量不大的场景能发现一些聚合逻辑的细节问题。我踩过的一个坑是窗口边界问题滚动窗口是左闭右开的比如[00:00, 00:05)00:05:00 整点的数据会落到下一个窗口。如果业务方理解成闭区间对账时就会差一条。这种问题一定要在联调阶段和业务方确认清楚。5.4 压力测试与性能瓶颈定位上线前必须做压力测试。我用的是自己写的压测脚本往消息队列里灌数据逐步加大 QPS观察 Flink 任务的各项指标指标正常范围异常表现可能原因吞吐量随 QPS 线性增长增长停滞算子成为瓶颈背压0 或低持续高背压下游处理慢Checkpoint 时长稳定逐渐变长状态膨胀CPU 使用率60% 以下持续 90%计算密集或 GCGC 时间低频繁 Full GC内存不足或泄漏背压是最直观的指标。Flink Web UI 里每个算子都有背压指示红色表示背压严重。如果过滤算子背压高说明黑名单查询慢如果窗口算子背压高说明聚合逻辑重或者状态大。我遇到过一次窗口算子背压排查后发现是allowedLateness设得太长10 分钟导致窗口状态一直不释放。改成 1 分钟后背压立刻消失。这个参数不是越大越好要根据数据迟到情况合理设置。6. 几个容易翻车的细节和我的应对经验6.1 黑名单数据的格式一致性黑名单从风控系统同步过来格式一定要统一。我遇到过风控系统有时候发 JSON、有时候发纯文本的情况导致广播流解析失败黑名单更新中断。解决办法是在广播流入口加一层格式校验解析失败的数据写到侧流告警不让它影响主流程。DataStreamString validatedBlacklist rawBlacklist .process(new ProcessFunctionString, String() { Override public void processElement(String value, Context ctx, CollectorString out) { try { // 尝试解析验证格式 JSON.parseObject(value); out.collect(value); } catch (Exception e) { // 格式错误输出到侧流 ctx.output(new OutputTagString(invalid-blacklist){}, value); } } });6.2 黑名单删除的同步问题黑名单不只有新增还有删除比如误判后解封。如果广播流只发新增删除操作就同步不过去。我的做法是广播流里带上操作类型ADD表示加入黑名单REMOVE表示移出。Override public void processBroadcastElement(BlacklistOp op, Context ctx, CollectorAdClickEvent out) throws Exception { BroadcastStateString, Boolean state ctx.getBroadcastState(descriptor); if (ADD.equals(op.getType())) { state.put(op.getUserId(), true); } else if (REMOVE.equals(op.getType())) { state.remove(op.getUserId()); } }这个细节很容易被忽略上线后才发现解封的用户还是被过滤业务方投诉才回头改。6.3 窗口结果输出的幂等性统计结果写到下游存储时要考虑幂等性。因为 Checkpoint 恢复后窗口可能重新触发同一个窗口的结果会输出两次。如果下游是数据库用INSERT ... ON DUPLICATE KEY UPDATE或者先删后插来保证幂等。如果下游是消息队列消费端要做去重。我一般会在结果里带上窗口的起止时间下游按广告位 窗口起始时间做主键重复写入直接覆盖。6.4 监控告警的配置任务上线不是终点监控才是。我必配的几个告警Checkpoint 连续失败超过 3 次就告警可能是状态太大或存储有问题。背压持续过高某个算子背压超过 5 分钟说明处理能力不足。黑名单过滤量为 0如果一段时间内过滤量突然归零可能是黑名单同步断了。任务重启非预期重启要立即通知排查是资源问题还是代码 bug。监控指标可以从 Flink 的 Metrics 系统导出到 Prometheus再用 Grafana 做面板。这套组合用起来很顺手社区里资料也多。6.5 资源规划的保守原则最后说资源规划。我见过太多任务因为资源给少了上线后频繁背压、Checkpoint 失败。我的原则是按峰值 QPS 的 1.5 倍规划资源并且留出至少 30% 的内存余量给状态和 GC。具体来说如果峰值 QPS 是 10 万单并行度能处理 1 万 QPS那至少开 15 个并行度。内存方面如果状态预估 2 GBTaskManager 内存至少给 4 GB留一半给堆外和 GC。这个配置不是拍脑袋是压测出来的经验值。资源给多了浪费给少了出事宁可稍微宽裕一点。广告点击量统计这个任务技术栈不复杂但细节特别多。黑名单过滤、窗口聚合、状态管理、容错恢复每一环都有坑。我把这几年的经验整理出来希望能帮你少走点弯路。实际落地时建议先在测试环境把链路跑通用模拟数据验证过滤和统计逻辑再做压力测试确定资源配置最后小流量上线观察一段时间。稳扎稳打比一步到位靠谱得多。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
