企业官网建设流程全解析

简介：基于Spark的实时用户画像分析系统是一份面向大数据工程师、数据平台架构师及推荐/营销算法团队的技术PDF，聚焦如何利用Spark生态构建面向海量用户的实时画像平台。内容以优酷用户画像系统为实例，完整讲解了系统架构、技术选型、筛选器设计、Join模型优化与存储方案，覆盖Spark、Hadoop、Scala、ANTLR、SQL等核心组件，并给出3~10亿用户、500G数据、50+维度、5000+标签下的性能Benchmark，能帮助读者理解从数据采集、标签计算到精准投放/推荐落地的全链路方法。资源为单个PDF文件，大小2.74MB，便于阅读和收藏。目前已有324人学习，适合希望系统掌握实时用户画像体系设计思路，并借鉴优酷工程实践经验的大数据从业者。

1. 实时用户画像的落地形态：Spark在这里解决什么问题

运营同学早上发来需求：给最近30分钟活跃但过去7天没下单的用户发一张优惠券。如果走T+1离线数仓，等标签算出来用户可能已经进入沉睡期；如果走自研流处理，又很难复用已有的数据清洗逻辑。实时用户画像系统的本质，是让“用户标签”从离线批量更新变成事件驱动更新，在秒级或分钟级完成标签写入。

Spark在这个场景里不是唯一选择，但它有一个很实际的优势：同一套DataFrame代码既可以跑批，也可以跑流。常见做法是用Spark Structured Streaming消费Kafka事件，借助内存计算和分布式执行，把ETL逻辑直接复用到实时链路。这个能力对团队很重要，因为画像的维度会随时间膨胀，离线开发一套逻辑、实时再开发一套，维护成本会翻倍。

下面的内容面向需要从0搭建或重构实时画像的数据工程师、数据平台工程师和沾数据后端的业务开发。你先能看到标签体系怎么建，再看到Spark实时ETL怎么写、集群和内存怎么调，最后给出线上质量校验和迭代技巧。全程都有可复现的代码和参数，照着落地基本不会跑偏。

2. Spark实时用户画像的标签体系与数据建模

实时用户画像不仅仅是“存一个用户维度表”，它决定你后续所有分析的边界。一个常见的错误是先定技术再补标签，导致Spark任务里堆满if else，却不知道标签的业务口径。我的习惯是先按业务需求拆标签维度，再回到Spark里做特征计算。

2.1 从业务需求反推标签维度

标签体系可以从更新频率和计算逻辑两个角度拆分。按计算逻辑分，最常用的是三类：规则型标签直接处理事件字段，统计型标签依赖窗口聚合，算法型标签来自模型评分。更新频率上还要区分状态快照与累计指标，例如“是否在直播间”只要存最后一次状态，“过去7天访问次数”则需要逐天累加。两者对实时写入路径的要求完全不同。下面这张表是我在实际项目里整理出的标签维度划分，重点看实时入口与更新频率的搭配。

这张表看着简单，但实际建模时有一个关键点：不同更新频率会影响Spark的存储选型。秒级状态标签要放Redis，分钟级统计标签适合放HBase或Doris，天级偏好标签可以留在数仓里。后端第3章会具体说明存储对接方式。设计阶段还必须输出标签字典，包含标签名、中文名、计算逻辑、更新频率、责任人。没有字典，画像跑一个月后，没人知道“高活跃用户”到底指什么。

2.2 用DataFrame做标签数据的清洗与加工

清楚了标签分类，下一件事就是把原始日志变成标准客户行为事件。这里我强烈推荐直接用PySpark DataFrame，而不是RDD。DataFrame有内置的schema推断和谓词下推，写出来的ETL脚本更像SQL，不同资历的工程师都能看懂。下面这个最小示例，同时演示了Kafka读取、schema解析和规则标签生成。

from pyspark.sql import SparkSession from pyspark.sql.functions import to_timestamp, when, col spark = SparkSession.builder \ .appName("user_profile_label_etl") \ .enableHiveSupport() \ .getOrCreate() # 读取Kafka中的埋点日志，这里以二进制格式为例 raw_df = spark.read.format("kafka") \ .option("kafka.bootstrap.servers", "kafka001:9092,kafka002:9092") \ .option("subscribe", "user_behavior") \ .load() # 解析value并筛选关键字段 event_df = raw_df.selectExpr("cast(value as string) as json_str") \ .select( from_json("json_str", "userId STRING, eventType STRING, timestamp STRING, itemId STRING") ) \ .select( col("userId"), col("eventType"), to_timestamp("timestamp", "yyyy-MM-dd HH:mm:ss").alias("event_time") ) # 生成规则标签：加购未支付、活跃用户、流失风险 labeled_df = event_df.withColumn( "label", when(col("eventType") == "add_cart", "cart_without_pay") .when(col("eventType") == "view", "active_user") .when(col("eventType") == "order", "buyer") .otherwise("unknown") ) labeled_df.write.format("parquet").mode("overwrite").save("/tmp/user_labels")

这个脚本的读流程很典型：从Kafka读取原始二进制消息，用from_json拆出userId、eventType等字段，然后通过withColumn做规则判断。注意from_json需要手写schema，在实时链路里很常见，因为流式数据不像离线表有统一元数据。实际ETL脚本还会把字段解析、事件清洗、标签计算拆成不同函数，便于Spark的物理计划复用。如果字段很多，建议用StructType而不是字符串拼schema，否则字段一旦嵌套多，维护会非常痛苦。

2.3 画像存储模型与宽窄表选择

标签计算完不是落Parquet就结束了，最终要供在线查询。离线数仓里我们习惯用宽表，一个用户一行，几十个字段。但实时画像场景不适合把全部标签塞进同一张宽表，原因有两个：一是标签更新频率不同，写宽表会导致频繁Update全行；二是Spark实时任务要保证低延迟，窄表按标签名写更容易做局部更新。我一般会把画像表设计成窄表：主键是user_id + tag_name，tag_value存放JSON或普通类型，update_time记录最后更新时间。在Redis里用hash结构存储，key是user_id，field是tag_name，value是tag_value。这一设计在启动Spark实时任务前就定好，后面集群调优、存储扩容才不会改模型。

窄表映射到Spark的分析路径也很自然。做用户群体分析时，可以通过Spark SQL把行转列，变成宽表供BI使用；在线查询时，同一份窄表又能按RowKey直接Get。这样实时链路和离线分析共用一份主模型，避免数据口径二次加工。

3. 用Structured Streaming把画像更新延迟降到秒级

第2章把标签模型和离线ETL脚本理清了，但离“实时”还有一步：如何让Spark持续消费事件，并把计算结果写到在线存储。这里明确说，不建议再用老牌的Spark Streaming基于DStream API，应该用Structured Streaming。

3.1 为什么选Structured Streaming：事件时间与状态管理

Structured Streaming把流表当无边界的DataFrame，你可以用同样的select、groupBy、agg操作来做实时计算。对于实时画像，最有价值的是水印和事件时间语义。埋点日志往往因为前端重试或网络抖动乱序，如果用处理时间计数，会得出“今日活跃下降”这种错误结论。Structured Streaming允许用watermark忽略迟到的数据，并在状态存储里保留中间结果。

3.2 实时画像ETL核心代码：窗口统计与标签合并

下面是一个典型的实时画像ETL框架，我整理成可以直接改的脚本。这个脚本会消费Kafka中的浏览和加购事件，按用户ID做5分钟滑动窗口统计，然后将结果upsert到具有主键的MySQL表中。这里用MySQL做示例，生产环境可以换成HBase或Redis。

from pyspark.sql.functions import window, col, count, when, to_timestamp, from_json stream_df = spark.readStream.format("kafka") \ .option("kafka.bootstrap.servers", "kafka001:9092") \ .option("subscribe", "user_behavior") \ .option("failOnDataLoss", "false") \ .load() \ .selectExpr("cast(value as string) as json") events = stream_df.select( from_json(col("json"), event_schema).alias("e") ).select( col("e.userId").alias("user_id"), col("e.eventType").alias("event_type"), to_timestamp(col("e.ts"), "yyyy-MM-dd HH:mm:ss").alias("event_time") ) # 窗口聚合计算行为标签 profile_updates = events \ .withWatermark("event_time", "5 minutes") \ .groupBy( window("event_time", "5 minutes", "1 minute"), "user_id" ).agg( count(when(col("event_type") == "view", 1)).alias("view_count"), count(when(col("event_type") == "add_cart", 1)).alias("cart_count") ) def write_upsert(df, epoch_id): df.write.format("jdbc") \ .option("url", "jdbc:mysql://profile-db:3306/profile") \ .option("dbtable", "user_realtime_label") \ .option("user", "profile") \ .option("password", "secret") \ .option("batchsize", "1000") \ .mode("append") \ .save() query = profile_updates.writeStream \ .foreachBatch(write_upsert) \ .outputMode("update") \ .trigger(processingTime="60 seconds") \ .queryName("user_profile_streaming") \ .checkpointLocation("/data/checkpoint/profile") \ .start() query.awaitTermination()

这里需要重点说明几个参数。withWatermark允许事件时间晚到5分钟，超过阈值的旧数据会直接丢弃，这是控制脏数据进入核心链路的关键。window定义5分钟窗口、1分钟滑动步长，保证标签口径是近5分钟。outputMode("update")只输出发生变化的结果，适合画像更新这种不断叠加的场景。foreachBatch则是一个万能出口，适合JDBC这类不支持流式写入的目标端，可以在批内做去重再写。checkpointLocation必须指向HDFS或云盘，不能放本地/tmp，否则任务重启时会丢offset。另外failOnDataLoss建议设false，Kafka提前删除消息时任务不会直接崩溃。

3.3 在线存储怎么选：Redis还是HBase

Spark实时任务计算完标签后需要写入在线系统。不同业务对读取延迟、更新频率要求不同。在我经历过的项目里，选型主要看两点：单个用户标签的读写并发，以及标签集合的大小。下面这个表格总结了常见选型。

如果并发量很高且要支持画像分析，可以选HBase作为离线结果与实时结果合并的存储层，再通过Phoenix或Presto做分析查询。之前设计的窄表模型映射到HBase的RowKey(user_id) + Qualifier(tag_name)非常自然。如果只是推荐系统取特征，Redis/hash结构更合适，但要记得给key设置TTL，否则长期用户的状态会撑爆内存。

4. Spark集群搭建与内存/ETL调优：让实时画像稳得住

实时任务不比离线夜批，出问题就得立即恢复，机器挂了三五分钟都肉眼可见。这一章把从Spark集群搭建到实时任务稳定运行的关键配置讲清楚。标题里的“集群搭建”对应这里，不只是启动master/worker，还要考虑提交模式与资源隔离。

4.1 最小Spark集群搭建与安装使用

真实生产环境用YARN/SBP返回调度，但很多团队会把“集群搭建”理解成用独立模式跑通，然后换YARN反而栽跟头。我建议先用Standalone模式跑通，再迁移到YARN，快速验证任务代码。下面的步骤以Spark 3.x为例，从Apache官网下载spark-3.x.x-bin-hadoop3.2.tgz，部署在3台机器上。

# 下载并解压到/data/soft tar -zxvf spark-3.x.x-bin-hadoop3.2.tgz -C /data/soft/ ln -s /data/soft/spark-3.x.x-bin-hadoop3.2 /data/soft/spark # 配置workers cd /data/soft/spark/conf cp spark-env.sh.template spark-env.sh echo "JAVA_HOME=/usr/java/jdk1.8.0_212" >> spark-env.sh echo "SPARK_MASTER_HOST=host01" >> spark-env.sh echo "SPARK_WORKER_CORES=8" >> spark-env.sh echo "SPARK_WORKER_MEMORY=24g" >> spark-env.sh echo "SPARK_WORKER_OPTS=-Dspark.worker.cleanup.enabled=true" >> spark-env.sh # 启动集群 /data/soft/spark/sbin/start-master.sh /data/soft/spark/sbin/start-workers.sh

这段命令的核心是SPARK_WORKER_MEMORY和SPARK_WORKER_CORES，它们决定这个节点能被多少executor共享。一个常见错误是把本机物理内存全配给worker，却不给操作系统和DataNode留余量，结果节点卡死。建议保留20%给系统。

4.2 实时任务的资源与内存参数设置

集群搭建完后，重点就变成Spark作业内的参数。实时画像任务采用spark-submit方式提交，通常我是这样设置执行参数的：

除了executor memory，还要注意spark.executor.memoryOverhead。YARN模式下默认是executor-memory的10%，如果任务频频报内存Overhead，多是因为Spark内部有大量网络缓冲和直接内存，建议手动设到1g以上。另外，Structured Streaming的checkpoint目录必须放在可靠的存储，比如HDFS，不能放本地/tmp。

4.3 ETL脚本中的常见性能瓶颈与spark-etl优化

实时画像的ETL脚本除了读Kafka，还会做维度关联、身份归一化等操作。最容易出现的性能问题是shuffle倾斜，个别user_id因为爬虫或者刷单流量巨大，导致某个任务节点长时间无法结束。有一组优化手段可以套用。使用随机前缀+两阶段聚合处理分钟级热点用户，在groupBy前把user_id打散成瞬时key；对维表用broadcast join，只要维表小于spark.sql.autoBroadcastJoinThreshold默认10MB，避免shuffle；关闭不必要的Spark UI日志保留策略，避免大段时间的driver overhead；使用DataFrame API而不是RDD，让Spark Catalyst优化器执行列剪裁和谓词下推。

在实时画像场景，最有效的是分区内限制记录数，可以在foreachBatch里先filter再写入，防止小文件爆炸。另一个容易忽略的点是：如果同一个Spark任务同时跑多个流，要让不同的queryName绑定不同的checkpoint路径，否则Kafka offset会相互覆盖。

5. 实时用户画像系统的质量校验与增量迭代技巧

系统上线后，不能只盯着延迟，还得回答“标签算得准不准”。我这里的做法是用离线Spark任务抽样对账，同时对晚到数据做标签回刷。

5.1 用离线任务校准实时标签

写一个离线任务，从数仓取同一时间段的用户行为明细，用相同的口径计算标签，然后和实时存储的标签做diff。

SELECT wr.user_id, wr.view_count AS realtime_cnt, off.view_count AS offline_cnt, IF(wr.view_count != off.view_count, 1, 0) AS diff_flag FROM (SELECT user_id, view_count from realtime_tag WHERE dt=#{predict_date}) wr JOIN (SELECT user_id, count(*) AS view_count from dwd_user_event WHERE event_time between #{start} and #{end} GROUP BY user_id) off ON wr.user_id = off.user_id HAVING diff_flag = 1 ORDER BY diff_cnt DESC;

需要注意：实时任务和离线任务取绝对相等意义不大，因为流处理窗口是滑动计算，离线按天切分会有边界数据。不要直接对diff_flag做全量告警，而应该设一个容忍比例，比如偏差超过5%才触发。

5.2 处理延迟数据与标签回刷

Structured Streaming的watermark会丢到5分钟后的数据，但这个值并不绝对。某些用户的行为轨迹可能延迟半小时，导致画像“卡住”。常见做法有两种：一是调大watermark到10-15分钟；另一种是写一个离线标签回刷脚本，扫描过去一天的事件日志，重新计算丢数据较多的标签。我比较常用第二种，因为watermark调大会增加状态存储成本。回刷脚本是一个Spark批任务，只需要在spark-submit时传入一个时间参数：

spark-submit --class com.example.RollbackLabel \ --master yarn \ --deploy-mode cluster \ --executor-memory 6g \ --executor-cores 3 \ /app/profile/tag-backfill.jar \ --begin 2025-03-01 --end 2025-03-01 --targetTag "view_count"

这个脚本以“T+1”的方式重新跑一遍用户行为，然后把计算出来的标签批量覆盖进Redis/HBase，这样第二天早晨数据就齐了。

5.3 用StreamingQueryListener做延迟监控与告警

比起写一堆自定义监控，我更建议直接使用Spark StreamingQueryListener，每5分钟记录一次进程的inputRowsPerSecond、processedRowsPerSecond、事件时间延迟等指标，有异常就直接推送到钉钉/企业微信。核心逻辑是在onQueryProgress回调里读progress对象，提取关键字段。下面是一个典型的progress输出：

{ "id": "user_profile_streaming", "name": "user_profile_streaming", "timestamp": "2025-03-01T10:00:00+08:00", "batchId": 15, "inputRowsPerSecond": 1000.0, "processedRowsPerSecond": 600.0, "durationMs": 5000 }

在实际工程里，我会把回调里拿到的processedRowsPerSecond写入InfluxDB，配合Grafana看趋势。如果连续3个微批processedRowsPerSecond低于正常值的50%，就触发告警。这个方式比观察Spark UI更及时，因为UI最多延迟几十秒，而告警可以直接打到值班群。

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