基于Flink流处理的亿级用户实时画像系统实战

简介本资源为基于Flink流处理的动态实时亿级全端用户画像系统完整项目包面向计算机、软件工程、人工智能等专业的在校学生与教师可用于毕业设计、课程设计、项目立项演示或进阶学习。项目围绕实时流计算与用户画像构建展开涵盖数据采集、标签计算、画像存储等核心环节适合具备一定Java与大数据基础的学习者参考。压缩包共327个文件以258个Java源码为主体辅以properties配置、xml与yaml环境定义、sql建表脚本、jar依赖、md说明文档及词典与图片等辅助资源整体约6.07MB结构完整、层次清晰。目前已有309人学习关注。代码经过测试运行成功配套数据集与详细文档可帮助读者快速理解Flink实时处理链路、画像标签体系与工程组织方式也可在此基础上修改扩展实现自定义功能适合作为毕设或课设的可靠参考。1. 从一份毕业设计说起Flink 流处理怎么撑起亿级用户画像电商大促的凌晨运营在后台圈了一批「近 30 分钟加购但未下单」的用户准备推券结果推送发出去转化率惨淡。排查发现这批标签是 T1 离线跑出来的用户昨晚加购、今早已经被别的活动转化过了。这就是静态画像的典型翻车现场——标签的时效性跟不上用户行为的节奏。要解决它就得把画像从「每天算一次」变成「行为发生即更新」而这正是 Flink 流处理的主场。这份「基于 Flink 流处理的动态实时亿级全端用户画像系统」的毕业设计核心就是干这件事把 App、小程序、H5、PC 全端的埋点行为接进来用 Flink 做实时聚合把用户标签算出来写进可查询的存储供运营和推荐实时调用。它适合三类人正在做大数据毕业设计、需要一套能跑通全链路的参考实现的同学刚接触 Flink 实时计算、想找一个完整项目练手的工程师以及手上已经有离线画像、想把它改造成实时链路的从业者。下面我按「数据怎么流、标签怎么算、坑在哪」把这条路走一遍。2. 全端埋点到实时标签链路怎么设计才不返工2.1 为什么是 Flink 而不是 Spark Streaming选型这件事毕业设计里最容易写成「因为 Flink 火」但真正落地时理由要具体。用户画像的实时计算有两个硬需求一是事件级低延迟用户点一下商品几秒内标签就要能反映出来二是状态要大一个用户可能积累几十上百个行为计数还要做去重、窗口聚合。Flink 的原生状态管理和事件时间语义在这两点上比微批模型更顺手。具体到画像场景Flink 的优势体现在三个地方。第一是 KeyedState按 userId 分组后每个用户的行为计数、最近一次行为时间这些都能放在托管状态里不用自己接 Redis 做中间存储。第二是事件时间加 Watermark埋点数据从端上上报到 Kafka 往往有乱序和延迟用事件时间窗口能保证「过去 5 分钟的行为」统计口径稳定不会因为到达顺序不同算出不同结果。第三是 Checkpoint 机制任务挂了能从上次快照恢复画像数据不会因为一次重启就全丢。Spark Streaming 不是不能用它的微批模型在秒级延迟场景下也够但状态 API 相对弱一些做复杂去重和会话窗口时要自己写不少逻辑。对于毕业设计这种要体现技术深度的项目Flink 的 DataStream API 和状态编程是更好的展示面。常见做法是埋点走 KafkaFlink 消费后做 ETL 清洗、窗口聚合、标签计算最后 sink 到 HBase 或 ClickHouse 供查询。2.2 分层架构ODS、DWD、DWS、ADS 各放什么一套能扩展的实时画像不该把所有逻辑塞进一个 Flink 作业。我一般会按数仓分层思路拆开这样每层职责清晰出问题也好定位。分层存放内容典型载体更新频率ODS原始埋点日志未清洗Kafka Topic实时写入DWD清洗后的事件明细补全维度Kafka / Hudi实时DWS按用户聚合的中间指标Flink 状态 / Redis实时ADS最终用户标签宽表HBase / ClickHouse实时/准实时ODS 层就是端上 SDK 上报的原始 JSON字段可能缺、可能有脏数据不要在这一层做任何业务计算。DWD 层做的是标准化统一时间格式、过滤测试账号、补上设备维度和渠道维度。DWS 层是画像的核心按 userId 聚合出「近 1 小时点击次数」「近 24 小时下单金额」这类中间指标。ADS 层才是运营能直接用的标签比如「高价值活跃用户」「流失预警用户」。这样分层的好处是当运营要加一个新标签时多数情况只需要在 ADS 层加一个计算逻辑复用 DWS 的中间指标不用动上游。毕业设计里如果只写一个 Job 从头算到尾答辩时被问「怎么扩展」会很难答。2.3 用 Flink DataStream 跑通第一个标签计算光说架构没用先跑一个最小可用的标签计算。假设 Kafka 里已经有埋点事件格式是{userId:u1,event:click,itemId:i100,ts:1700000000000}我们要算每个用户近 5 分钟的点击次数。// 1. 环境准备开启 Checkpoint保证状态可恢复 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.enableCheckpointing(60000); // 每 60 秒做一次快照 env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); // 2. 接入 Kafka 埋点数据 Properties props new Properties(); props.setProperty(bootstrap.servers, kafka:9092); props.setProperty(group.id, user-profile-job); KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(kafka:9092) .setTopics(ods_user_event) .setGroupId(user-profile-job) .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStreamString rawStream env.fromSource(source, WatermarkStrategy.noWatermarks(), kafka-source); // 3. 解析 JSON提取 userId、event、ts DataStreamUserEvent events rawStream .map(new MapFunctionString, UserEvent() { Override public UserEvent map(String value) throws Exception { JSONObject obj JSON.parseObject(value); return new UserEvent( obj.getString(userId), obj.getString(event), obj.getLongValue(ts) ); } }) .filter(e - e.userId ! null click.equals(e.event)); // 只保留点击事件 // 4. 分配 Watermark允许 5 秒乱序 DataStreamUserEvent withWatermark events.assignTimestampsAndWatermarks( WatermarkStrategy.UserEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) - event.ts) ); // 5. 按 userId 分组开 5 分钟滚动窗口计数 DataStreamTuple2String, Long clickCount withWatermark .keyBy(e - e.userId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .aggregate(new CountAggregator()); clickCount.print(); // 生产环境换成 sink 到 HBase/ClickHouse env.execute(user-profile-click-count);这段代码有几个参数必须说清楚。enableCheckpointing(60000)里的 60000 是毫秒代表每分钟做一次状态快照间隔太短会增加存储压力太长则故障恢复时丢的数据多画像场景一般 30 秒到 2 分钟都合理。forBoundedOutOfOrderness(Duration.ofSeconds(5))是 Watermark 的乱序容忍度设成 5 秒意味着窗口会等 5 秒再触发端上上报延迟大就调大但调太大会让标签更新变慢。TumblingEventTimeWindows.of(Time.minutes(5))是滚动窗口窗口之间不重叠适合做「近 5 分钟」这种固定周期统计如果要算「最近一次行为后 30 分钟内的转化」就得换成会话窗口。CountAggregator是一个自定义的 AggregateFunction用增量聚合而不是把窗口内所有元素攒起来再算内存占用小很多。这是画像场景的关键优化点亿级用户下全量缓存窗口数据会直接把 TaskManager 撑爆。3. 标签计算与状态管理亿级用户下怎么不 OOM3.1 KeyedState 存什么、怎么设 TTL画像标签分两类一类是累计型比如「历史总下单金额」需要长期保留另一类是滑动型比如「近 7 天活跃天数」过期就该清掉。如果所有状态都永久保留亿级用户跑几个月状态会膨胀到几百 GBCheckpoint 时间越来越长最后任务直接卡死。Flink 的 State TTL 就是解决这个的。给状态配一个过期时间超时未更新的 key 会被自动清理。// 定义带 TTL 的状态描述符 StateTtlConfig ttlConfig StateTtlConfig .newBuilder(Time.days(7)) // 7 天未更新则过期 .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // 写入时刷新过期时间 .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) // 不返回已过期数据 .cleanupInRocksDBCompactFilter(1000) // RocksDB 压缩时清理每 1000 条检查一次 .build(); ValueStateDescriptorLong lastActiveDesc new ValueStateDescriptor(lastActiveTime, Long.class); lastActiveDesc.enableTimeToLive(ttlConfig); ValueStateLong lastActiveState getRuntimeContext().getState(lastActiveDesc);Time.days(7)要和业务口径对齐如果标签定义是「近 7 天活跃」TTL 就设 7 天设短了标签会突然归零设长了状态清理不及时。setUpdateType选OnCreateAndWrite表示每次写入都刷新过期时间适合活跃用户如果选OnReadAndWrite读操作也会刷新可能导致冷用户状态一直不释放。cleanupInRocksDBCompactFilter是 RocksDB 后端下的清理策略全量遍历清理太耗性能放在压缩阶段顺带做更划算。这里有个血泪经验TTL 配置改了之后已经存在的状态不会立即按新规则清理需要等一次全量 Checkpoint 或者从 savepoint 重启才生效。上线前一定要在预发环境验证状态大小曲线别等生产环境 Checkpoint 超时了才发现。3.2 用 ProcessFunction 做标签的实时更新窗口聚合只能算固定周期的指标但画像里很多标签是「事件驱动」的——用户一下单就要立刻更新他的「最近下单时间」和「累计下单次数」。这种场景用KeyedProcessFunction更合适它能对每条事件做处理还能注册定时器。public class OrderTagFunction extends KeyedProcessFunctionString, OrderEvent, UserTag { private ValueStateLong orderCountState; private ValueStateLong lastOrderTimeState; Override public void open(Configuration parameters) { // 状态初始化配 TTL ValueStateDescriptorLong countDesc new ValueStateDescriptor(orderCount, Long.class); countDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(30)).build()); orderCountState getRuntimeContext().getState(countDesc); ValueStateDescriptorLong timeDesc new ValueStateDescriptor(lastOrderTime, Long.class); timeDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(30)).build()); lastOrderTimeState getRuntimeContext().getState(timeDesc); } Override public void processElement(OrderEvent order, Context ctx, CollectorUserTag out) throws Exception { Long count orderCountState.value(); count (count null ? 0L : count) 1; orderCountState.update(count); lastOrderTimeState.update(order.ts); // 实时输出标签下游 sink 到 HBase out.collect(new UserTag(order.userId, order_count_30d, String.valueOf(count))); out.collect(new UserTag(order.userId, last_order_time, String.valueOf(order.ts))); // 注册 24 小时后的定时器用于计算「是否流失」 ctx.timerService().registerEventTimeTimer(order.ts 24 * 3600 * 1000L); } Override public void onTimer(long timestamp, OnTimerContext ctx, CollectorUserTag out) throws Exception { // 定时器触发时检查用户是否还有新订单 Long lastOrder lastOrderTimeState.value(); if (lastOrder ! null lastOrder 24 * 3600 * 1000L timestamp) { out.collect(new UserTag(ctx.getCurrentKey(), is_churn_risk, true)); } } }processElement里每来一条订单就更新状态并输出标签这是实时性的来源。registerEventTimeTimer注册的定时器会在事件时间到达指定时刻时触发onTimer用来做「24 小时无下单则标记流失风险」这类逻辑。注意定时器也是状态数量多了同样占内存不要给每个事件都注册长周期定时器能合并的尽量合并。open方法里给状态配了 30 天 TTL因为「近 30 天下单次数」这个标签的定义就是 30 天窗口。如果业务改成「近 90 天」TTL 和标签口径要同步改否则算出来的数对不上。3.3 状态后端选 RocksDB 还是 HashMap状态后端的选择直接决定能扛多大数据量。HashMapStateBackend 把状态放在 JVM 堆内存里读写快但受限于堆大小几千万 key 就可能 OOM。RocksDBStateBackend 把状态存在本地磁盘内存里只放热数据能支撑亿级 key代价是读写有序列化开销延迟比纯内存高。画像系统我一般直接上 RocksDB因为用户量摆在那堆内存根本放不下。配置上要注意几点state.backend.rocksdb.memory.managed设为 true 让 Flink 统一管理内存state.backend.incremental设为 true 开启增量 Checkpoint否则每次全量快照几百 GB 根本传不动本地磁盘要用 SSD机械盘做 RocksDB 压缩会拖垮整个任务。# flink-conf.yaml 关键配置 state.backend: rocksdb state.backend.incremental: true state.backend.rocksdb.memory.managed: true state.checkpoints.dir: hdfs:///flink/checkpoints state.backend.rocksdb.localdir: /data/flink/rocksdbstate.checkpoints.dir指向 HDFS 或对象存储本地目录只放 RocksDB 的 SST 文件。localdir要选容量大、IO 好的盘Checkpoint 期间会有大量临时文件。如果发现 Checkpoint 频繁超时先看是不是本地磁盘写满了这是最常见的翻车原因。4. 避坑与排查这套链路最容易崩在哪4.1 数据倾斜导致个别 TaskManager 被打满现象Flink UI 上大部分 subtask 早就处理完了就一两个 subtask 的 records 数一直涨Checkpoint 卡在某个算子不推进。原因通常是 keyBy 的 key 分布不均比如某个大 V 用户或者测试账号产生了海量事件全落到同一个 subtask。解决办法分两步。先定位在 Flink UI 的算子详情里看各 subtask 的 records 和 bytes 分布明显偏高的就是热点 key。再处理如果是脏数据在 DWD 层直接过滤掉如果是真实热点做两阶段聚合——第一次给 key 加随机前缀打散聚合后再去掉前缀做第二次聚合。// 两阶段聚合打散热点 key DataStreamUserEvent salted events.map(e - { int salt ThreadLocalRandom.current().nextInt(10); // 0-9 随机前缀 return new UserEvent(salt _ e.userId, e.event, e.ts); }); // 第一次聚合按加盐后的 key 算局部结果 // 第二次聚合去掉盐按真实 userId 合并加盐的粒度要试10 个前缀通常够用太多会增加第二次聚合的负担。这个方案会增加一次网络 shuffle但能换来负载均衡热点严重时值得。4.2 Checkpoint 超时或失败现象Job 运行一段时间后 Checkpoint 一直处于 IN_PROGRESS最后超时失败任务反复重启。原因可能有三类状态太大导致快照慢、下游 sink 反压导致 barrier 对齐慢、HDFS 或对象存储写入慢。排查顺序先看 Checkpoint 的 duration 和 size 趋势如果 size 持续增长说明状态没清理干净检查 TTL 是否生效如果 size 正常但 duration 长看反压指标多半是 sink 端写入慢如果都正常检查存储的带宽和权限。# 查看 Checkpoint 历史定位失败原因 flink list -m yarn-cluster -yid application_xxx # 在 Flink UI 的 Checkpoint 页面看每个 subtask 的 snapshot 耗时一个容易被忽略的点state.backend.rocksdb.localdir如果配在系统盘Checkpoint 期间大量写会拖慢整个节点甚至影响其他任务。生产环境一定单独挂数据盘。4.3 事件时间与处理时间混用导致标签口径错乱现象运营反馈「近 1 小时活跃用户数」忽高忽低和实际对不上。原因往往是代码里一部分算子用了事件时间另一部分用了处理时间或者 Watermark 生成策略配错。排查方法检查所有涉及时间的算子确认assignTimestampsAndWatermarks在 keyBy 之前调用且窗口用的是EventTimeWindows而不是ProcessingTimeWindows。如果埋点数据本身时间戳有问题比如端上时钟不准事件时间会失真这时要么在 DWD 层做时间校正要么退而用处理时间但接受口径偏差。提示上线前用一批带已知时间戳的测试数据跑一遍人工核对窗口输出比看日志靠谱。4.4 Sink 到 HBase 时连接数暴涨现象任务跑起来后 HBase RegionServer 报连接数超限Flink 任务频繁抛异常。原因是每条记录都新建一个 Connection没有复用。解决在RichSinkFunction的open方法里初始化连接invoke里复用close里关闭。同时配连接池和批量提交别一条一条写。public class HBaseSink extends RichSinkFunctionUserTag { private Connection conn; private BufferedMutator mutator; Override public void open(Configuration parameters) throws Exception { Configuration conf HBaseConfiguration.create(); conn ConnectionFactory.createConnection(conf); BufferedMutatorParams params new BufferedMutatorParams(TableName.valueOf(user_tag)) .writeBufferSize(2 * 1024 * 1024); // 2MB 缓冲 mutator conn.getBufferedMutator(params); } Override public void invoke(UserTag tag, Context context) throws Exception { Put put new Put(Bytes.toBytes(tag.userId)); put.addColumn(Bytes.toBytes(info), Bytes.toBytes(tag.tagName), Bytes.toBytes(tag.tagValue)); mutator.mutate(put); // 进缓冲区异步批量提交 } Override public void close() throws Exception { if (mutator ! null) mutator.close(); if (conn ! null) conn.close(); } }writeBufferSize设 2MB 是个折中太小批量效果差太大故障时丢的数据多。mutator.mutate是异步的不会每条都阻塞吞吐能上去。如果对一致性要求高可以在 Checkpoint 时调mutator.flush()确保数据落盘。4.5 毕业设计里最容易忽略的数据源造假现象本地跑得好好的一上集群就没数据或者数据量对不上。很多毕业设计的数据集是离线文件用readTextFile读但实时链路要求数据是流式进来的。如果直接把历史日志文件当流读Watermark 会瞬间推到未来窗口全部触发算出来的标签毫无意义。正确做法要么用 Kafka 真实接入要么写一个模拟器按时间戳逐条发送。模拟器里要控制发送速率别一股脑全推出去。# 模拟埋点上报按真实时间间隔发送 import json, time, random from kafka import KafkaProducer producer KafkaProducer(bootstrap_serverskafka:9092, value_serializerlambda v: json.dumps(v).encode()) users [fu{i} for i in range(1000)] events [click, view, order, cart] while True: event { userId: random.choice(users), event: random.choice(events), itemId: fi{random.randint(1, 500)}, ts: int(time.time() * 1000) } producer.send(ods_user_event, event) time.sleep(0.01) # 控制速率约 100 条/秒time.sleep(0.01)决定了模拟速率压测时可以调小演示时调大让数据看起来更真实。ts用当前时间戳保证事件时间和处理时间接近窗口行为符合预期。5. 让画像真正可用查询侧优化与一个验证技巧标签算出来只是第一步运营和推荐系统要能毫秒级查到才行。HBase 按 userId 做 RowKey 查询很快但运营经常要按标签反查人群比如「找出所有近 7 天活跃且客单价大于 200 的用户」这种多条件组合查询 HBase 就力不从心了。我的做法是双写HBase 存明细供点查ClickHouse 存标签宽表供圈人。ClickHouse 的列式存储和向量化执行对这种多条件过滤加聚合的场景非常合适。写 ClickHouse 时注意用ReplacingMergeTree引擎同一个 userId 的标签更新会覆盖旧值查询时加FINAL或者用argMax取最新。表结构按标签类型分列别用 EAV 模型一行一个标签那样查询要反复 join性能差一个数量级。CREATE TABLE user_tag_wide ( user_id String, last_active_time DateTime, order_count_30d UInt32, order_amount_30d Decimal(18,2), is_churn_risk UInt8, update_time DateTime ) ENGINE ReplacingMergeTree(update_time) ORDER BY user_id;ORDER BY user_id是排序键点查和按用户范围扫描都快。ReplacingMergeTree(update_time)保证同一 user_id 保留 update_time 最大的那条。如果要按标签圈人可以再建物化视图或者投影把常用标签组合做成索引。验证这套链路是否真的实时我有个笨但有效的办法准备一个测试账号手动触发一次下单事件然后用秒表计时看 ClickHouse 里这个账号的order_count_30d多久加一。正常应该在 3 到 10 秒内如果超过 30 秒说明链路上有环节在攒批或者 Watermark 设太大。这个测试比看监控面板直观得多答辩时演示也很有说服力。最后说个我踩过的坑早期为了追求「实时」把所有标签都做成秒级更新结果下游 HBase 写入压力巨大RegionServer 频繁 compaction查询反而变慢。后来把标签分了级——行为类标签秒级更新统计类标签分钟级更新画像类标签小时级更新——整体吞吐降了一半查询延迟反而更稳。实时不是越快越好匹配业务节奏才是对的。希望帮到你。本文还有配套的精品资源点击获取