Spark Structured Streaming实时用户画像系统实战：从架构选型到Redis落库

简介：《基于Spark的实时用户画像分析系统》是一份技术分享型PDF，面向大数据工程师、数据分析师及推荐系统开发者，讲解如何利用Spark构建实时用户画像平台，解决精准营销与个性化推荐中的用户行为理解问题。资源包为1个PDF文件、2.74MB，内容覆盖系统架构、技术栈、功能特点、性能优化与应用场景。文档以优酷用户画像系统为实例，重点展示高效筛选器与Join模型优化，包括Code Generator、内存列式存储、Bitmap压缩等手段，在3~10亿用户、500G数据量、50多个画像维度、5000多个标签的规模下，将筛选响应时间控制在2秒，群体合并10~20秒，对比分析15~20秒。这些真实Benchmark数据对设计高性能实时画像系统极具参考价值。目前已有325人学习下载。适合需要了解用户画像系统整体设计、Spark实时计算实践以及大数据交互式分析优化思路的读者，兼具框架讲解与落地细节。

1. 为什么"等明天"的画像让业务慢半拍：Spark实时画像系统的定位与第一道选型题

用户画像分析系统大家都不陌生——你打开App看到的推荐、营销活动选人、风控规则命中，背后都是画像标签在干活。传统做法是T+1离线跑批，昨天产生的行为今天才落到标签上。这套逻辑在日活几百万、业务决策按天走的时代没问题，但到了实时调度、实时营销、实时风控这些场景，等一天就是等竞争对手把用户抢走。我下面要拆解的这套基于Spark的实时用户画像分析系统，核心思路是跳出离线批处理思维，用Spark Structured Streaming消费行为事件流，在秒级到分钟级延迟内产出标签并写进特征存储，供在线服务查询。适合谁看？正在做实时数仓、实时特征服务，或者被离线画像延迟坑过的数据平台工程师，这篇是把这套系统讲清楚、能落地的一篇实战笔记。

2. 实时画像架构选型与标签分层：Lambda、Kappa与流批一体怎么选，标签体系怎么设计

实时画像要解决的第一件事不是写代码，而是把架构和标签模型定下来。很多团队上来就写Structured Streaming，写到一半发现离线链路和实时链路口径对不上，标签重复建设，最后越改越乱。动手前把下面两个问题想清楚，后面能少返工一半。

2.1 实时画像不是把离线任务加速跑：从Lambda到Kappa的架构取舍

先明确一个观点：实时画像不是把离线Spark任务缩短运行周期，而是换一条数据流转路径。离线任务每天凌晨读Hive，算完后覆盖前一天的全量标签；实时任务则持续消费Kafka里的行为事件，在事件发生的当下或几十秒内完成统计。两条链路的代码、计算模型、数据源形态完全不同。

当前主流的架构选择有三条路，我整理成一张对比表：

我见过的大部分公司最终都落在"简化版Lambda"上：离线标签继续保留，做兜底和对账；实时标签只覆盖高时效场景，比如实时营销触达、实时风控、实时调度。Kappa架构在公司里推行阻力很大，原因很现实——Kafka保留全量事件几个月的存储成本，比想象中高得多，而且状态管理出问题时排查难度不小。

选型时还有两个前置条件容易被忽略。一是Spark集群本身的资源水位，实时任务和离线任务共用集群，如果晚上离线大任务把资源吃满，实时任务的延迟就会抖动；二是埋点数据质量，实时画像的底料是事件流，如果应用侧埋点缺失、重复上报，实时链路会比离线更容易把脏数据暴露给下游。常见的做法是实时任务跑在独立队列，或者至少给实时任务打上单独的调度标签。

2.2 标签体系怎么设计：事实标签、规则标签与模型标签的分层逻辑

把架构定了，下一步是设计标签体系。我习惯把标签分成三层：事实标签、规则标签、模型标签。这三层对应不同的计算频率和数据来源，不能混在一个Spark任务里算。

事实标签是最底层，直接来自行为事件。比如"最近5分钟浏览商品数""今日下单总金额"，这类标签用Structured Streaming的窗口聚合就能算出来，数据源单一，逻辑简单，实时性最强。规则标签是在事实标签上做业务规则判断，比如"近10分钟内加购超过3次的用户标记为高意向用户"，这类标签需要把事实标签组合起来，状态管理更重，但规则本身经常被运营调整。模型标签靠机器学习打分，比如"用户未来24小时购买概率"，这类标签依赖模型推理，一般在离线程算出基础分，再在实时链路里根据最新行为做增量修正。

设计时有三条原则我踩过坑后才真正理解。第一，事实标签要尽量原子化，一个标签只表达一个事实，便于上层复用；第二，规则标签的规则要配置化，运营改规则时不应该让开发改代码重新发布；第三，模型标签不要试图完全实时化，异步更新往往性价比更高。拿一个农产品行情App举例，用户今天搜了三次"西兰花"是事实标签，"生鲜价格敏感用户"是规则标签，"明天会再次打开App的概率"是模型标签，三者时效要求完全不同。

这也解释了为什么实时数仓开发工作内容里，很大一部分精力花在标签分层和口径治理上，而不是写聚合逻辑。标签分层没想清楚，后面做规则配置和特征服务时，会不断面临"这个标签该谁算、算完存哪、多久过期"的重复讨论。记住一点：实时画像系统里最值钱的不是流处理技术，而是标签模型的复用能力。

3. 用Spark Structured Streaming搭建实时画像管道：Kafka接入、窗口聚合与状态管理参数

架构和标签模型定好后，就可以动手写管道了。这一章我按三步走：先把Kafka事件读进来并解析JSON，再做窗口聚合算出事实标签，最后在输出阶段用Spark SQL算规则标签。全程用PySpark实现，Spark集群搭建好之后这套脚本可以直接在yarn集群上提交运行。

3.1 Spark集群就绪后的第一件事：Kafka接入与JSON解析的PySpark脚本

大多数公司的事件流统一进Kafka，实时画像的第一步就是把Kafka里的用户行为事件读成DataFrame。这里有个容易忽略的点：Kafka里的value是二进制，必须先用from_json解析，而且schema要提前和埋点侧对齐。

from pyspark.sql import SparkSession from pyspark.sql.functions import from_json, col, from_unixtime from pyspark.sql.types import (StructType, StructField, StringType, LongType, DoubleType) # 构建SparkSession，shuffle分区数默认200，流任务建议显式调小 spark = SparkSession.builder \ .appName("realtime-user-profile") \ .config("spark.sql.shuffle.partitions", "8") \ .config("spark.sql.streaming.schemaInference", "true") \ .getOrCreate() spark.sparkContext.setLogLevel("WARN") # 事件schema必须与埋点字段严格对齐，新增字段要按兼容规则演进 event_schema = StructType([ StructField("user_id", StringType()), StructField("event_type", StringType()), # view / click / buy / cart StructField("item_id", StringType()), StructField("category_id", StringType()), StructField("price", DoubleType()), StructField("ts", LongType()) # 毫秒时间戳，来自服务端 ]) # 消费Kafka事件流，startingOffsets用latest保证只取新事件 raw = spark.readStream \ .format("kafka") \ .option("kafka.bootstrap.servers", "kafka-1:9092,kafka-2:9092") \ .option("subscribe", "user_behavior_events") \ .option("startingOffsets", "latest") \ .load() # 先from_json解析成结构体，再展开到顶层列 events = raw.select( from_json(col("value").cast("string"), event_schema).alias("event") ).select("event.*") \ .withColumn("event_time", from_unixtime((col("ts") / 1000).cast("long")).cast("timestamp"))

这段代码里值得注意的参数有三个。spark.sql.shuffle.partitions默认200，离线任务无所谓，但流任务每个微批都会触发shuffle，200个分区会拖慢小数据量的处理，我一般在8到16之间调。startingOffsets用latest意味着任务启动后只消费新事件，如果业务上需要补算重启前的数据，改成earliest。from_unixtime((col("ts") / 1000).cast("long"))这行是把毫秒时间戳转成秒再生成TimestampType，直接除以1000会被推断成DoubleType，时间就会算错，这是读取JSON时最容易翻车的一个细节。

3.2 窗口聚合与状态管理：30秒窗口内行为标签的计算代码与参数

事件读进来之后，要做的是按用户维度做窗口聚合。Structured Streaming里的窗口聚合是有状态的，系统会为每个窗口保存中间结果，watermark用来决定状态什么时候可以清理。

from pyspark.sql.functions import window, count, sum, when # watermark设60秒，允许事件晚到60秒内仍参与计算 events_with_watermark = events \ .withWatermark("event_time", "60 seconds") # 每30秒滑动一次，统计每个用户在最近60秒的浏览/点击/加购/购买行为 behavior_stats = events_with_watermark \ .groupBy( col("user_id"), window(col("event_time"), "60 seconds", "30 seconds") ) \ .agg( count(when(col("event_type") == "view", 1)).alias("view_cnt"), count(when(col("event_type") == "click", 1)).alias("click_cnt"), count(when(col("event_type") == "buy", 1)).alias("buy_cnt"), sum(when(col("event_type") == "buy", col("price"))).alias("buy_amount") ) \ .select( col("user_id"), col("window.start").alias("window_start"), col("window.end").alias("window_end"), col("view_cnt"), col("click_cnt"), col("buy_cnt"), col("buy_amount") )

窗口参数.window(..., "60 seconds", "30 seconds")，第一个参数是窗口长度，第二个是滑动步长。每30秒算一次最近60秒的聚合，意味着同一个事件会落在两个窗口里，好处是标签更平滑，坏处是计算量翻倍。如果业务上不需要平滑，直接把两个参数都设成30秒，就是滚动窗口，每30秒只算一个窗口。watermark设60秒，我建议按事件的最大乱序程度来定，不是拍脑袋。

3.3 规则标签用Spark SQL维护：把运营规则写成SQL而不是Java代码

事实标签算出来后，下一步是叠加规则标签。这里的关键是规则不能写在Java/Scala/Python代码里，否则运营每次调规则都要改代码重新提交Streaming任务。我一般把事实标签注册成临时视图，然后用Spark SQL写规则。

def apply_rule_tags_to_redis(batch_df, batch_id): """每个微批先算规则标签，再和事实标签一起写入Redis""" if batch_df.isEmpty(): return batch_df.createOrReplaceTempView("behavior_stats") rule_tags = spark.sql(""" SELECT user_id, 'high_intent' AS tag_name, CASE WHEN buy_cnt >= 3 OR click_cnt >= 10 THEN '1' ELSE '0' END AS tag_value, window_end FROM behavior_stats """) # 这里把rule_tags与事实标签合并后写Redis，下一章展开

用Spark SQL管理规则的好处是规则的可读性大幅提升，运营同学能直接看懂"买过3次或点过10次就是高意向用户"，而不是猜代码逻辑。而且SQL天然支持版本管理，把SQL文本存进Git，每次改动都有历史记录。规则标签的实时性取决于窗口聚合的粒度，实际业务里"高意向"的判定窗口往往是5到15分钟，60秒窗口只是示例参数，要根据场景调。

4. 实时特征落库与在线查询：Redis在画像服务里的数据结构选型与写入参数

Spark把标签算出来后，如果没有一个在线存储让业务服务毫秒级拿到标签，这套系统就白做了。实时特征服务的核心诉求是低延迟、高并发、支持过期——这三个词基本把存储选型指向了Redis。

4.1 为什么Redis是实时画像的默认在线存储：对比HBase与ClickHouse的点查能力

先看一张选型对比，这是我被问得最多的一个问题：

用户画像的点查模式是按user_id取一批标签，Redis的Hash结构完美匹配这个场景：key是profile:user:{user_id}，field是标签名，value是标签值。一次hgetall就能拿到一个用户的全部实时标签，RTT只有一次。HBase在数据量上更强，能存PB级画像数据，但一个在线实时画像系统如果只靠点查，几百GB内存的Redis集群足够覆盖几亿活跃用户的高频标签。ClickHouse更适合分析师跑画像人群SQL，不适合在线服务调用。

选Redis还有一个隐性的便利：TTL机制天然契合实时标签的时效。离线画像标签一存就是一个月，实时画像标签可能30秒后就过期了。Redis的过期策略让数据自动淘汰，不需要额外写清理任务，这在实时数仓开发里是一个很大的省心事。

4.2 标签写入与毫秒级读取：foreachBatch加Pipeline的落地代码与连接池参数

具体写入我推荐用foreachBatch而不是foreach。foreachBatch每个微批调用一次，可以复用Redis连接、合并写入、在批内再聚合；foreach是每条记录调用一次，只适合写入量极小的场景。下面是完整的写入代码。

from redis import Redis # decode_responses=True 让hget返回str而不是bytes，省去手动解码 r = Redis(host="redis-feature", port=6379, db=0, decode_responses=True, socket_connect_timeout=3) def write_tags_to_redis(batch_df, batch_id): """每个微批将事实标签和规则标签批量写入Redis""" if batch_df.isEmpty(): return batch_df.createOrReplaceTempView("behavior_stats") rule_tags = spark.sql(""" SELECT user_id, 'high_intent' AS tag_name, CASE WHEN buy_cnt >= 3 OR click_cnt >= 10 THEN '1' ELSE '0' END AS tag_value, window_end FROM behavior_stats """) # 用pipeline批量发送，减少网络RTT，transaction关闭避免MULTI/EXEC开销 pipe = r.pipeline(transaction=False) for row in rule_tags.collect(): key = f"profile:user:{row.user_id}" pipe.hset(key, mapping={ "view_cnt_60s": 1, # 实际应从事实标签透传 "high_intent": row.tag_value, "updated_at": row.window_end.strftime("%Y-%m-%d %H:%M:%S") }) pipe.expire(key, 60) # 标签60秒过期，与窗口时长保持一致 pipe.execute() query = behavior_stats.writeStream \ .foreachBatch(write_tags_to_redis) \ .outputMode("update") \ .trigger(processingTime="30 seconds") \ .option("checkpointLocation", "hdfs:///user/spark/checkpoint/profile-v1") \ .start() query.awaitTermination()

这段代码要重点理解三个参数。pipeline(transaction=False)把一批命令打包发送，命令数多时吞吐能提升好几倍，关掉事务避免MULTI/EXEC的额外开销。expire(key, 60)让标签在60秒后自动过期，下游服务读不到过期标签时再回退到离线兜底值，这是实时画像和离线画像衔接的好设计。checkpointLocation必须放在HDFS或S3这类共享存储上，如果放在本地磁盘，任务重启后状态就丢了，这个坑下面具体说。

在线服务读取侧的代码简单得多，但有一个参数值得注意：要用连接池，不能在每次请求里新建连接。

import redis from redis.connection import ConnectionPool # 在线服务用连接池复用连接，max_connections按峰值QPS估 pool = ConnectionPool(host="redis-feature", port=6379, db=0, decode_responses=True, max_connections=50) r = redis.Redis(connection_pool=pool) # 一次RTT拿到用户全部实时标签 tags = r.hgetall("profile:user:9527")

读取侧真正的坑在decode_responses。如果读写两侧有一侧没开这个参数，hgetall会返回一堆bytes对象，到代码里还得逐个decode，小问题但很烦。另外，给Redis配置maxmemory-policy allkeys-lru能保证内存写满时优先淘汰最久没访问的标签，这个策略对画像这种"热标签反复读、冷标签被淘汰"的场景非常合适，这是我用Redis存画像标签时最常调的一个参数。

5. 实时画像的5个坑与排查：乱序、状态膨胀、热点倾斜、checkpoint、Redis背压

实时画像和离线画像最大的差别在于：离线任务跑挂了可以重跑，实时任务跑挂了会丢数据、会延迟。这一章我用几个真实场景梳理最常遇到的坑，每个问题按"现象、原因、解决"的顺序讲，排查思路可以直接套用。

5.1 事件乱序让标签算错：Watermark设60秒还是5分钟

现象：用户行为标签在某些时段异常偏低，比如晚间App端上报的事件明显少了，但离线日志显示用户行为很活跃。排查发现是用户设备网络状态差异导致事件到达Kafka的时间不统一，4G弱网下的事件可能延迟几分钟才到，而窗口聚合早就关闭了。

原因：Structured Streaming按事件时间做窗口，但默认情况下事件晚到会被直接丢弃，watermark决定了允许晚到多久。watermark设太小，晚到事件全部丢失；设太大，状态保存时间拉长，内存压力变大。

解决：watermark的设置不能凭感觉，我一般先看事件延迟分布。用Kafka consumer的records-lag-max和埋点里的客户端时间戳做对比，统计P95延迟。如果95%的事件在90秒内到达，watermark设120秒到180秒比较稳。"多丢一点数据还是多撑一会内存"是个权衡，纯实时营销场景我倾向设大一点，因为标签算错比延迟更伤害业务；对账场景才需要严格控制watermark。

5.2 状态膨胀拖垮executor：状态清理与Spark内存配置的血泪经验

现象：任务刚上线时一切正常，跑了一周后executor频繁Full GC，任务处理延迟从30秒恶化到5分钟，最终OOM。

原因：Structured Streaming的状态存储会为每个窗口、每个key保存聚合中间值。窗口长度大、key数量多、watermark设得久，状态就会持续膨胀。尤其是用户规模大但窗口又不合理的任务，状态增长速度远超预期。

解决：三个手段要一起用。第一，缩窗口——如果业务允许，把窗口从1小时改成15分钟，状态量直接降四分之三；第二，收紧watermark——晚到容忍度从10分钟降到3分钟，状态清理会更激进；第三，给executor配off-heap内存，状态存的是Java对象，堆内存不够时会触发GC风暴，在提交参数里加spark.memory.offHeap.enabled=true和spark.memory.offHeap.size=2g能缓解。还有一个偏方：把计算拆成两个任务，一个算窗口内的增量，一个单独做状态清理，见过有人这么干，但维护成本高，不推荐。

5.3 热点用户造成数据倾斜：加盐聚合在流任务里的正确姿势

现象：同一个Streaming任务里，大部分executor都空闲，只有一个executor跑满CPU。查看Spark UI发现某个task的输入数据量是其他task的上百倍，这是一个超级用户产生的行为事件占了大头。

原因：groupBy(user_id)在shuffle时按user_id哈希到固定分区，热点用户的所有事件都落进同一个分区，分区内的任务自然成为瓶颈。

解决：离线批处理里的加盐思路在流处理里要谨慎用。正确的做法是：事件写入Kafka时就按user_id % shard_num分片，保证同一个用户有序落盘，聚合时先按分片聚合，再做第二层合并。

-- 第一段聚合：按分片键和窗口先算出局部结果 SELECT shard_id, user_id, window_start, window_end, COUNT(*) AS cnt FROM events_with_shard GROUP BY shard_id, user_id, window_start, window_end; -- 第二段聚合：合并分片结果 SELECT user_id, window_start, window_end, SUM(cnt) AS total_cnt FROM stage1 GROUP BY user_id, window_start, window_end;

但也要说清楚：这种"加盐"方式只对普通热点有效，如果某个超级用户的量级是普通用户的上百万倍，拆到16个shard还是单台executor扛。超大用户的终极方案是单独识别、单独走一条高配置链路，也就是热点隔离，特征服务里对头部用户单独建桶，这是各大厂的标准做法。

5.4 checkpoint恢复失败：升级之后任务起不来的后悔药

现象：集群的Spark版本从3.3升级到3.4，同时调整了事件schema，加上了一个字段。重启Streaming任务后，直接报CheckpointMissingException或序列化错误，任务起不来。

原因：Structured Streaming的checkpoint里保存了状态存储的schema和操作符元数据，Spark小版本升级或计算逻辑变更都可能导致checkpoint与当前代码不兼容。Structured Streaming官方明确不保证跨版本恢复，这个问题遇到过一次就长记性了。

解决：我的习惯是——每次升级或者改schema之前，先把checkpoint目录完整备份一份到另一个路径；如果代码逻辑大变，直接放弃旧checkpoint，用一个新appName和新checkpoint目录重新跑，必要时用startingOffsets: earliest从Kafka最早偏移消费历史数据补算。还有一点，checkpoint目录一定不能用回收站，HDFS的trash回收策略可能在任务没结束就清掉老版本的状态，这个坑我踩过一次后，给checkpoint目录单独配了禁用trash的权限。

5.5 Redis写入成为瓶颈：消费延迟排查与背压处理

现象：Kafka消费延迟持续上涨，但看Spark UI发现executor的CPU使用率不高，GC也不频繁。查Streaming指标发现每个微批的处理时间越来越长，从3秒涨到30秒。

原因：foreachBatch里的Redis操作拖慢了整个微批。如果没用pipeline而是一条条写Redis，每条都要走一次网络RTT，一个微批几万条记录，光写Redis就要几十秒，Kafka端自然堆积。

解决：第一步把逐条写入改成pipeline批量写入，能解决大部分问题；第二步关闭transaction=True，因为事务在pipeline里会触发MULTI/EXEC，额外的往返让性能回到逐条水平；第三步如果写入量实在太大，考虑换架构——把Redis写入从流任务里拆出去，流任务只写到Kafka sink，再用独立的消费者服务通过异步批量写Redis。拆链路会增加维护成本，但吞吐瓶颈从Spark转到Redis时，这是最通用的解法。

6. 画像准不准、值不值：延迟监控、抽样比对与AB验证方法

实时画像系统上线不等于结束，真正的考验是证明标签准、延迟低、业务有价值。这一章给三个验证手段，都是我每个新标签上线必做的事。

6.1 标签准确性验证：和离线画像对账的抽样方法

实时标签跑得再快，如果算错了，下游业务就是在错误数据上做决策。我每个新标签上线前都要先做对账：每天凌晨用离线任务重算一遍同样的标签，再和实时标签做对比。抽样方法很关键，不能只比对活跃用户，要按用户分段分层抽样，头部用户、中部用户、尾部用户各抽一部分，避免都被头部用户的数据淹没。

对账的核心指标是"标签一致率"，两个链路算出来的标签值在允许误差范围内的比例。不同的标签类型对账口径不一样，事实标签比的是数值误差，规则标签比的是命中率，模型标签比的是排序相关性。做一个简单的对账表：

6.2 实时标签的性价比排序：先做哪些标签，后做哪些标签

实时计算很贵，不是所有标签都值得实时化。我给标签做优先级排序时只看一个维度：标签晚一小时拿到，业务损失是多少。实时营销触达、实时风控拦截这几个场景，晚一小时等于钱没花出去或者坏账多了一笔，必须实时；推荐系统粗排里用户兴趣标签，晚15分钟影响可以接受；人群洞察、用户分层这类运营分析标签，走离线就够了，不需要占用实时资源。

我自己的习惯是，每次上线一个新标签，先写对账SQL再写规则表达式，标签上线前先在离线表上跑一遍历史数据分布，确认口径无误，再提交Streaming任务。这套流程多花半天时间，但能省下后面排查口径不一致的好几个通宵。实时画像从来不是"上了一个流任务"就结束的事，把验证机制和标签分层做好，系统才真正能跑稳、跑久。希望帮到你。

本文还有配套的精品资源，点击获取