1. 广告点击统计场景的核心需求拆解广告点击量统计这件事听起来像是数据平台里最基础的一个指标计算但真正落到生产环境里坑远比想象中多。我做过好几个广告归因和反作弊相关的实时计算项目几乎每一个都是从“简单统计一下点击量”开始最后演变成一套包含黑名单过滤、维度聚合、异常检测的完整链路。这个项目标题里的“黑名单过滤”四个字恰恰是整个统计链路中最容易被低估、却最影响数据质量的一环。先说清楚这个项目要解决什么问题。广告投放系统里每一次广告曝光和点击都会产生一条日志日志里包含用户标识、广告位标识、时间戳、渠道来源等字段。运营侧需要实时看到每个广告位、每个渠道的点击量用来判断投放效果、调整预算分配。但原始点击日志里混入了大量无效点击同一用户短时间内的重复点击、被标记为异常的设备产生的点击、来自特定渠道的机器行为等等。如果不做过滤直接统计出来的点击量会虚高运营拿着失真的数据做决策轻则浪费预算重则误判整个投放策略。所以这个项目的本质是用 Flink 做实时流式聚合在聚合之前先经过一层黑名单过滤把已知的无效点击剔除掉再按广告位和渠道维度统计点击量。适合谁来参考我认为有三类人值得细看一是刚接触 Flink 流处理、想找一个完整可运行案例的开发者二是已经在做实时统计、但数据质量一直被业务方质疑的工程师三是需要理解“过滤逻辑放在哪个环节”这个架构决策的技术负责人。关键词里提到的“黑名单过滤”在 Flink 里实现方式有很多种可以是一条独立的过滤算子也可以是一个 BroadcastStream 动态下发的规则还可以是维表 Join 的一种变体。选哪种取决于黑名单的更新频率、数据量和一致性要求。这个项目我倾向于用广播状态的方式来做原因后面会详细展开。2. 整体架构设计与方案选型思路2.1 为什么选择 Flink 而不是其他流处理框架广告点击日志的特点是数据量大、延迟要求高、统计维度相对固定但需要支持动态调整。我实测过一个中等规模的广告系统高峰期每秒的点击日志在几万到十几万条之间用传统的定时批处理跑延迟至少是分钟级运营侧根本等不了。Flink 的优势在于它的事件时间语义和状态管理机制能够保证在乱序数据到达的情况下依然算出准确的窗口统计结果。具体来说广告点击日志从产生到进入计算链路中间会经过采集、传输、缓冲等多个环节到达顺序和实际发生顺序往往不一致。比如用户先点了广告 A 再点广告 B但 B 的日志可能先到。如果用处理时间做窗口这两个点击可能被分到不同窗口里统计结果就偏了。Flink 的事件时间加上 Watermark 机制可以容忍一定程度的乱序等数据基本到齐了再触发窗口计算结果更准确。另一个关键点是状态。黑名单过滤需要维护一份“哪些用户/设备需要被过滤”的名单这份名单可能很大而且需要动态更新。Flink 的 KeyedState 和 BroadcastState 提供了原生的状态管理能力不用自己在外围维护一套缓存系统减少了架构复杂度。2.2 黑名单过滤的三种实现路径对比在 Flink 里做黑名单过滤我实际用过三种方式各有适用场景实现方式适用场景优点缺点算子内硬编码/配置文件加载黑名单极小且几乎不变实现简单无额外依赖更新需重启作业无法动态调整维表 Join如查外部存储黑名单大、更新频繁灵活支持大规模名单每条数据都要查外部存储延迟高、压力大广播状态BroadcastState黑名单中等规模、需要动态更新名单缓存在内存过滤无外部调用支持动态下发名单不能太大受限于 TaskManager 内存这个项目我选择广播状态方案。理由很直接广告黑名单通常不会大到几千万条一般几十万到几百万条量级完全放得进内存同时运营侧需要能够实时增删黑名单条目广播状态正好支持从一条控制流动态更新规则。维表 Join 的方式虽然也能动态更新但每条点击日志都要去查一次外部存储在每秒十万级 QPS 下外部存储的压力会非常大而且网络往返带来的延迟会拖累整个作业的吞吐。广播状态的机制是这样的一条流是主数据流也就是广告点击日志另一条流是规则流也就是黑名单的增删事件。规则流被广播到所有并行子任务每个子任务在本地维护一份黑名单副本。当规则流有新事件到达时更新本地副本主数据流的数据到达时直接查本地副本做过滤。这样过滤操作是纯内存的速度极快而且规则更新也能实时生效。2.3 统计维度的设计考量点击量统计的维度不是随便定的需要和业务方的看数习惯对齐。我见过不少项目一开始只按广告位统计后来业务方要按渠道拆、再后来要按时间段拆、再再后来要按用户分群拆每次加维度都要改代码、重跑历史数据非常被动。这个项目里我预设了两个核心维度广告位 ID和渠道 ID。广告位 ID 是广告投放的基本单元渠道 ID 标识流量来源。这两个维度组合起来基本能覆盖运营侧 80% 的看数需求。如果后续要加维度比如按地域、按设备类型可以在 KeyBy 的时候把新维度加进去窗口逻辑不用大改。窗口的选择上我用的是滚动窗口窗口长度 1 分钟。为什么不用滑动窗口因为滑动窗口会产生重复计算同一个点击会被统计到多个窗口里对于“点击量”这种需要精确计数的指标来说不合适。滚动窗口每个数据只属于一个窗口结果可以直接累加。1 分钟的长度是权衡了实时性和吞吐之后的选择太短了窗口数量多、状态压力大太长了运营侧看到数据的延迟就高。3. 核心细节解析与实操要点3.1 事件时间与 Watermark 的配置细节事件时间处理是 Flink 流计算里最容易出问题的地方我踩过的坑至少有一半和它有关。广告点击日志里的时间戳字段通常是从日志本身解析出来的但要注意时区问题。如果日志里的时间是 UTC而你的窗口逻辑按东八区来算统计结果就会差 8 小时。我的做法是统一在解析阶段把时间戳转成毫秒级 epoch 时间后续所有计算都用这个值避免时区混乱。Watermark 的生成策略我一般用BoundedOutOfOrdernessTimestampExtractor允许的最大乱序时间设为 5 秒。这个值怎么定我的经验是看数据链路的实际乱序程度。可以在接入端先做一个采样统计看看同一批数据里最大时间差是多少。如果链路比较稳定5 秒足够如果经过多层缓冲、乱序严重可能要放到 10 秒甚至 30 秒。但也不能太大Watermark 延迟越大窗口触发越晚端到端延迟就越高。注意Watermark 的乱序容忍时间和窗口长度要匹配。如果窗口是 1 分钟乱序容忍设了 2 分钟那窗口要等到 3 分钟后才触发实时性就没了。一般乱序容忍不超过窗口长度的 10% 到 20%。3.2 广播状态的初始化与更新逻辑广播状态的使用有一个关键点两条流的连接顺序和状态描述符的定义。我见过不少人在BroadcastProcessFunction里把processElement和processBroadcastElement的逻辑写反导致主数据流的数据被当成规则处理或者规则被当成数据过滤结果完全不对。正确的做法是主数据流调用connect方法连接广播流然后process里传入一个BroadcastProcessFunction的实现。在这个实现里processElement处理主数据流的每条点击日志processBroadcastElement处理规则流的每条黑名单更新事件。状态描述符用MapStateDescriptor定义key 是黑名单条目的标识比如用户 ID 或设备 IDvalue 可以是简单的布尔值或者一个包含生效时间、来源等信息的对象。规则流的更新逻辑要区分“新增”和“删除”。新增就是把 key 放进 MapState删除就是从 MapState 里移除。这里有个细节如果规则流里同时有新增和删除事件要保证顺序性。我一般会在规则事件里带一个操作类型字段processBroadcastElement里根据操作类型分别处理。3.3 黑名单匹配的字段选择与性能考量黑名单匹配用哪个字段直接决定了过滤效果和性能。常见的匹配字段有用户 ID、设备 ID、IP 地址、Cookie ID。用户 ID 最准确但很多场景下点击日志里没有登录态拿不到用户 ID。设备 ID 相对稳定但存在设备重置、模拟器伪造的问题。IP 地址容易误伤一个公司出口 IP 下可能有几百个正常用户。我的做法是优先用设备 ID 匹配设备 ID 为空时降级用 IP 匹配。这样既保证了大部分场景的准确性又不会因为字段缺失导致过滤失效。在广播状态里我会把设备 ID 和 IP 分别作为 key 存两份或者用一个复合 key。复合 key 的好处是一次查找就能判断坏处是内存占用翻倍。如果黑名单规模不大复合 key 完全可行如果规模很大就要考虑分开存储、分别匹配。性能方面广播状态的查找是 O(1) 的哈希查找单次过滤的开销可以忽略不计。真正的瓶颈在于状态的大小和序列化。如果黑名单有上百万条每个 TaskManager 都要存一份内存压力不小。我一般会监控 TaskManager 的堆内存使用率超过 70% 就要考虑优化比如把黑名单条目做压缩、或者改用维表 Join 方案。4. 实操过程与核心环节实现4.1 环境准备与依赖配置这个项目用 Maven 管理依赖核心是 Flink 的 streaming-java 和 flink-connector-kafka。版本选择上我建议用 Flink 1.14 或以上因为广播状态的 API 在 1.14 之后稳定了很多。Kafka 连接器用和 Flink 版本匹配的版本不要混用。dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version1.14.4/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version1.14.4/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version1.14.4/version /dependency如果要用事件时间还需要flink-streaming-java里的 Watermark 相关类这些都在标准依赖里不用额外引入。4.2 数据源接入与日志解析广告点击日志从 Kafka 接入topic 假设叫ad-click-log。日志格式我用 JSON字段包括user_id、device_id、ip、ad_slot_id、channel_id、click_time。解析的时候用 Jackson 或者 FastJSON我习惯用 Jackson因为它的流式解析在大量数据下性能更稳。DataStreamAdClickEvent clickStream env .addSource(new FlinkKafkaConsumer(ad-click-log, new SimpleStringSchema(), properties)) .map(new AdClickLogParser()) .assignTimestampsAndWatermarks( WatermarkStrategy.AdClickEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) - event.getClickTime()) );AdClickLogParser里要做几件事JSON 解析、字段校验、时间戳转换。字段校验很重要我遇到过日志里ad_slot_id为空的情况如果不校验直接往下传KeyBy 的时候会报空指针。我的做法是解析失败或者关键字段缺失的日志直接丢弃同时打一个计数器方便监控数据质量。4.3 黑名单规则流的构建规则流也从 Kafka 接入topic 叫blacklist-rule。规则事件的格式我设计成{op: ADD/DELETE, type: DEVICE/IP, value: xxx}。op标识操作类型type标识黑名单类型value是具体的值。DataStreamBlacklistRule ruleStream env .addSource(new FlinkKafkaConsumer(blacklist-rule, new SimpleStringSchema(), properties)) .map(new BlacklistRuleParser()); MapStateDescriptorString, Boolean blacklistDescriptor new MapStateDescriptor(blacklist-state, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.BOOLEAN_TYPE_INFO); BroadcastStreamBlacklistRule broadcastRuleStream ruleStream.broadcast(blacklistDescriptor);这里用MapStateDescriptor定义状态key 是黑名单值value 用 Boolean 表示是否存在。实际项目中如果还需要记录生效时间、来源等信息可以把 value 换成一个自定义的 POJO但要注意 POJO 必须可序列化。4.4 广播状态过滤与窗口聚合的串联过滤和聚合的串联是整个作业的核心。我用一个BroadcastProcessFunction做过滤输出过滤后的点击流然后再接 KeyBy 和窗口。SingleOutputStreamOperatorAdClickEvent filteredStream clickStream .connect(broadcastRuleStream) .process(new BlacklistFilterFunction()); DataStreamAdClickCount resultStream filteredStream .keyBy(event - event.getAdSlotId() _ event.getChannelId()) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new ClickCountAggregator(), new ClickCountWindowFunction());BlacklistFilterFunction里processElement方法查本地黑名单状态如果命中就丢弃否则collect输出。processBroadcastElement方法根据规则事件更新状态。这里有个关键点广播状态在processElement里是只读的不能修改这是 Flink 的设计约束目的是保证状态一致性。窗口聚合用aggregate而不是apply因为aggregate是增量聚合每来一条数据就更新一次累加器性能比apply好很多。ClickCountAggregator里维护一个计数ClickCountWindowFunction在窗口触发时把计数和窗口信息一起输出。4.5 结果输出与下游对接统计结果输出到 Kafka 的ad-click-counttopic格式是 JSON包含广告位 ID、渠道 ID、窗口开始时间、窗口结束时间、点击量。下游可以接实时大屏、告警系统或者写入 OLAP 数据库供查询。resultStream .map(new AdClickCountToJson()) .addSink(new FlinkKafkaProducer(ad-click-count, new SimpleStringSchema(), properties));输出的时候要注意窗口时间的格式。Flink 的窗口时间戳是 epoch 毫秒直接输出可读性差我一般会转成yyyy-MM-dd HH:mm:ss格式。另外窗口结束时间是窗口的右边界实际统计的是[start, end)区间内的数据这个语义要在输出里体现清楚避免下游误解。5. 常见问题与排查技巧实录5.1 黑名单过滤不生效的几种原因过滤不生效是最常见的问题我排查过多次原因基本集中在三类第一类是规则流没有正确广播。检查broadcast方法是否调用、状态描述符是否和processBroadcastElement里用的一致。我遇到过一次两个地方用了不同的描述符名字编译不报错但状态对不上规则更新完全没效果。第二类是匹配字段不一致。比如规则里存的是设备 ID但过滤时查的是用户 ID自然匹配不上。排查方法是打印几条规则和几条点击日志人工比对字段值。第三类是状态更新延迟。广播状态的更新是异步的规则流的数据到达后需要一点时间才能同步到所有子任务。如果规则刚下发就立刻有数据进来可能过滤不及时。这个延迟通常在毫秒级对大部分场景可以接受但如果业务要求强一致就要考虑其他方案。5.2 窗口不触发或结果延迟的排查窗口不触发九成和 Watermark 有关。检查 Watermark 是否正常生成、是否推进到了窗口结束时间之后。可以在 Flink Web UI 的 Watermark 指标里看当前 Watermark 值如果一直不涨说明数据流里没有新数据或者时间戳解析有问题。另一个常见原因是数据倾斜。如果某个广告位的点击量特别大对应的 KeyBy 子任务压力就会很大窗口触发也会慢。我一般会先看 Flink UI 里各子任务的吞吐和延迟如果明显不均就要考虑加盐或者预聚合。5.3 状态过大导致内存溢出的处理广播状态每个子任务存一份黑名单大了内存就吃紧。我遇到过黑名单涨到 500 万条时TaskManager 频繁 Full GC作业几乎不可用。解决办法有几个一是把黑名单按类型拆分设备 ID 和 IP 分开存减少单份状态的大小二是用RocksDBStateBackend替代内存状态后端把状态放到磁盘上代价是访问速度慢一些三是定期清理过期规则比如只保留最近 30 天生效的黑名单。提示RocksDB 状态后端对广播状态的支持有限广播状态默认还是存在内存里。如果黑名单实在太大建议改用维表 Join 方案把黑名单放到外部存储用异步 IO 查询。5.4 常见问题速查表问题现象可能原因排查方法解决措施过滤不生效规则流未广播/描述符不一致检查 broadcast 调用和描述符统一描述符确认广播逻辑窗口不触发Watermark 未推进查看 Watermark 指标检查时间戳解析和乱序容忍结果延迟高乱序容忍过大/数据倾斜看端到端延迟和子任务负载调小乱序容忍加盐处理倾斜内存溢出黑名单过大监控堆内存和 GC拆分状态、用 RocksDB、清理过期规则统计结果偏大黑名单未覆盖/匹配字段错抽样比对过滤前后数据补充黑名单修正匹配字段6. 实操心得与后续扩展方向这个项目跑通之后我在实际运维中又陆续加了一些东西这里分享几个我觉得比较有价值的扩展点。第一个是过滤命中率的监控。光看统计结果不够还要知道黑名单过滤掉了多少数据。我在BlacklistFilterFunction里加了两个计数器一个统计总数据量一个统计被过滤的数据量通过 Flink 的 Metrics 系统暴露出来。这样能直观看到黑名单的效果如果命中率突然飙升可能是遇到了异常流量需要及时排查。第二个是黑名单的自动更新。手动维护黑名单效率太低我后来接了一个离线任务每天跑一次异常检测把识别出的可疑设备 ID 自动写入规则流。这样黑名单就能自我迭代不用人工干预。第三个是多级过滤。广播状态过滤只能处理已知的黑名单对于未知的异常行为无能为力。我在后面又加了一层基于规则的过滤比如“同一设备 1 秒内点击超过 10 次”就判定为异常。这层过滤用 KeyedProcessFunction 实现维护每个设备的点击计数和上次点击时间逻辑不复杂但效果很好。最后说一个我踩过的坑广播状态的清理。黑名单里的条目如果只增不删时间长了状态会无限膨胀。我一开始没注意这个问题跑了两个月后发现状态大小涨到了几个 G。后来加了一个定时清理逻辑把超过 90 天没有更新的黑名单条目自动移除状态大小才稳定下来。这个清理逻辑可以放在规则流里由上游控制也可以在 Flink 作业里起一个定时器定期扫描状态并删除过期条目。这个项目后续还可以往实时告警方向扩展比如点击量突增突降时触发告警或者和广告预算系统联动点击量超过阈值时自动暂停投放。这些扩展的核心逻辑都不复杂关键是把过滤和统计的基础打牢后面的功能就是在这个基础上叠加。