企业官网建设流程全解析

简介：基于Spark的电商用户画像数据挖掘项目源码，面向大数据开发与数据挖掘学习者，聚焦电商平台用户行为分析、偏好建模与精准营销场景。项目采用Scala与Java为主，结合Spark分布式计算框架，覆盖用户标签体系、RFM模型、ETL数据处理及前端可视化展示等核心环节。压缩包共462个文件，大小13.45MB，包含296个class文件、70个scala源文件、20个java源文件，以及properties、xml、json等配置文件和JavaScript、CSS、HTML等前端资源，模块划分清晰，便于按tags-model、tags-etl、tags-web等目录研读。已有339人学习下载，适合希望掌握Spark实战应用、了解电商用户画像落地流程的开发者参考。通过源码可梳理从数据清洗到标签建模、再到结果展示的完整链路，为自建画像系统提供可复用的设计思路与代码基础。

1. 基于 Spark 的电商用户画像数据挖掘项目源码：先看清这套东西在解决什么

把“基于 Spark 的电商用户画像数据挖掘项目源码”这串标题拆开看，核心是三件事：标签体系怎么设计、Spark 在链路里管哪几段、代码以什么结构交付。很多刚转数据挖掘的工程师，拿到这类项目第一反应是去看算法，但真正让画像项目跑不动的，往往是 ID 混乱、口径漂移、宽表设计不合理这类离算法很远的问题。下面按标签分层、特征加工、源码结构、提交命令、踩坑排查和上线验证的顺序，把一套能落地的电商用户画像方案讲清楚，适合正在搭离线画像标签体系、想读源码却不知道从哪下手的读者。

2. 先把画像的数据立起来：标签分层与 ID-Mapping 的落地设计

用户画像不是一个结果表，而是一套带时间戳的标签集合。今天下单的活跃用户，三个月没回访后就不再是“高价值客户”；昨天刚注册的用户，还没有消费行为却已经有了设备 ID。所以第一件事不是写算法，而是把数据分层和用户 ID 统一这两个地基打好。地基不稳，后面 RFM 算得再漂亮，分群结果也没人敢信。

2.1 四层数据架构：ODS、DWD、标签明细层与 ADS 应用层

常见做法是把画像相关数据分成四层，每一层解决一个明确问题。

ODS 层表结构尽量跟上游保持一致，JSON 日志就原样存 JSON，订单快照就原样存全字段，这一层不做业务判断。DWD 层是工作量最大的地方，要处理字段缺失、时间格式不统一、重复订单这类脏数据，输出是标准化后的列式表，推荐用 Parquet 存储。TAG 层才是真正意义上的“用户画像”，它把用户的基础属性、消费能力、活跃度、品类偏好揉成一张宽表。ADS 层直接给运营和推荐系统用，通常是分群结果或者聚合指标。

为什么一定要分四层而不是一张表打天下？因为标签的生命周期不同。基础属性标签半年都不变，消费行为标签每天都要更新，模型分每周重算一次，混在一张表里会让调度和回溯都很难受。四层各管各的节奏，上游重跑时下游可以单独刷新，这是离线数仓里最稳的节奏。

2.2 ID-Mapping：把手机号、设备 ID、open_id 合并成唯一 user_id

电商场景里同一个用户可能同时拥有注册 ID、手机号、设备 ID、微信 open_id。行为日志里记的是 device_id，订单表里是注册 user_id，小程序里又是 open_id。如果不先把这些 ID 映射到同一个主键，后面所有 join 都会漏数据，而且漏得毫无规律。ID-Mapping 的输出是一张dim_user_id_mapping表，逻辑上可以这样建：

-- 为每个来源ID绑定一个唯一的 user_id -- 一次登录/绑定行为里出现的多个ID，归到同一个用户 insert overwrite table dim_user_id_mapping select id_value, id_type, min(user_id) as user_id from ( select phone as id_value, 'phone' as id_type, user_id from ods_user_bind_log where phone is not null and phone <> '' union all select device_id as id_value, 'device' as id_type, user_id from ods_user_bind_log where device_id is not null and device_id <> '' ) t group by id_value, id_type;

这个 SQL 的逻辑是：把每一条绑定记录里的手机号和设备 ID 分别拆成一行，然后按 ID 本身做聚合，取最小的 user_id 作为该 ID 的归属用户。参数上要注意两点：id_type必须显式声明，避免手机号和设备 ID 撞值；min(user_id)只是启发式策略，生产环境里用户 ID 存在合并、解绑、再绑定场景，常见做法是用连通分量或者 Union-Find 算法来重算，而不是简单取最小。这张表建好后，还要建一个反向索引，也就是给每个 user_id 保留全部历史 ID，方便排查“这个订单为什么归到那个用户”。

ID 映射表不是建一次就完事的，绑定关系会变：用户换手机、注销后重新注册、老账号并号。所以我一般会让这张表每天增量更新，同时保留first_seen_dt字段，方便追溯一个 ID 第一次出现的时间。如果某天发现用户量异常波动，先查这张表是不是绑定关系被上游改了，大概率问题出在这里。

3. 用 Spark SQL 做特征加工：从订单、日志到标签宽表的组装

标签宽表是整个画像项目里最核心的一张表，所有下游分群和模型都从它取数。特征加工的工作，用 Spark SQL 完成是非常顺手的：它天然支持大表 join、窗口函数和复杂聚合，比写 MR 任务高效得多，也比单机 DataFrame 能扛的数据量大几个数量级。这一章把从数据源接入到宽表组装的链路拆开讲。

3.1 数据源盘点：会员、订单、日志、商品四类数据怎么接

一个典型的电商用户画像项目，数据源大概有四类。

接入顺序有讲究。会员资料和商品维度表是小表，直接全量拉，不要做增量，量不大而且变更频率低。订单表是大表，按分区读取，每天一个分区，按ds过滤。行为日志是典型的半结构化数据，用 Spark 的 JSON 读取能力直接解析，但必须配合 schema 声明，不能依赖 schema 推断。接入完成后第一件事是确认主键和更新时间字段：订单表的主键是order_id加user_id，日志表的主键是user_id加event_time加page_id，主键不明确，后面去重无从下手。

3.2 清洗规则：去重、字段修正与时间窗口截断

接入之后先清洗，再做聚合。清洗行为日志时，常见做法是预先定义好 schema，然后用 Spark 的 DataFrame API 做处理：

import org.apache.spark.sql.types._ val behaviorSchema = StructType(Seq( StructField("user_id", StringType, true), StructField("device_id", StringType, true), StructField("page_id", StringType, true), StructField("action", StringType, true), StructField("event_time", LongType, true), StructField("extra", MapType(StringType, StringType), true) )) val rawDF = spark.read .schema(behaviorSchema) .json(rawPath) val cleanDF = rawDF .filter(col("user_id").isNotNull && length(trim(col("user_id"))) > 0) .withColumn("event_dt", from_unixtime(col("event_time") / 1000, "yyyy-MM-dd")) .filter(col("event_dt").between(beginDt, endDt)) .dropDuplicates("user_id", "event_time", "page_id", "action")

先声明 schema 再读，可以避免 Spark 对 JSON 字段类型反复推断导致的性能损耗。event_time是毫秒时间戳，所以除以 1000 转成秒再格式化。dropDuplicates比distinct精确：distinct会去整行，但行为日志里两条记录可能只有extra字段不同，实际是同一动作的重复上报，按业务主键去重才有效。between截断时间窗口也很关键，画像任务的起始日期不能写死，必须由调度参数传入，不然补数的时候会漏掉历史数据。

3.3 标签宽表组装：三次 left join 的先后顺序

宽表的基本单位是“一个用户一行”，所有标签字段拼成一列。组装顺序我一般固定为：先算事实标签（订单聚合、活跃度），再算偏好标签（品类集中度），最后 join 小表补充维度属性。

// 订单聚合：近90天消费频次、金额、最近购买天数 val orderAgg = spark.sql(s""" select user_id, count(distinct order_id) as frequency_cnt, sum(paid_amount) as monetary_amt, datediff(current_date(), max(pay_time)) as recency_days from dwd_order_clean where ds = '${ds}' and pay_time >= date_sub(current_date(), 90) group by user_id """) // 品类偏好：近90天购买次数最多的前3个类目 val preferAgg = spark.sql(s""" select user_id, concat_ws(',', collect_list(category_name)) as category_pref from ( select user_id, category_name, row_number() over(partition by user_id order by cnt desc) as rn from ( select user_id, category_name, count(1) as cnt from dwd_order_clean where ds = '${ds}' and pay_time >= date_sub(current_date(), 90) group by user_id, category_name ) t1 ) t2 where rn <= 3 group by user_id """) val wideTable = baseDF .join(orderAgg, Seq("user_id"), "left") .join(preferAgg, Seq("user_id"), "left") .withColumn("active_level", when(col("frequency_cnt").isNull, "new_user") .when(col("frequency_cnt") >= 5, "high") .otherwise("normal"))

这里两个子查询都提前做 group by，把订单明细从亿级行压缩到百万级用户行，再 join 宽表，shuffle 量小很多。preferAgg里的窗口函数row_number用来取每个用户购买次数前三的类目，concat_ws把它们拼成一个字符串，方便下游直接读。active_level是对空值的处理：没有订单的用户标记为新用户，而不是直接填空字符串，这样下游分群时不需要再做一层空值判断。所有 join 的关联键统一使用user_id，绝对不要混用 ID 类型，这是宽表组装里最容易翻车的地方。

4. 源码跑通的最小闭环：目录结构、核心表与提交命令

拿到一套画像项目源码，先不要急着读算法类，先看目录结构和提交脚本。源码组织方式决定了这项目能不能被团队接手。常见的 Spark 画像工程会按调用链拆成四个目录：bin、sql、etl、model，再加一个 pom.xml 或 build.sbt。

4.1 源码目录：bin、sql、etl、model 怎么编排

user-profile/ ├── bin/ │ ├── init_schema.sh │ ├── submit_daily.sh │ └── rerun_daily.sh ├── conf/ │ └── env.conf ├── sql/ │ ├── dwd/ │ │ ├── 01_behavior_clean.sql │ │ └── 02_order_clean.sql │ ├── tag/ │ │ ├── 01_rfm_score.sql │ │ └── 02_user_tag_wide.sql │ └── ads/ │ ├── 01_user_segmentation.sql │ └── 02_seg_daily_snapshot.sql ├── etl/ │ └── UserProfilePipeline.scala ├── model/ │ ├── RFMScorer.scala │ └── KMeansSegmenter.scala └── pom.xml

SQL 目录按分层命名，文件名带序号表示执行顺序，这一点很重要。etl目录只有一个 Scala 入口类，负责串联流程：启动 SparkSession、按依赖顺序执行 SQL、最后把宽表写入目标分区。model目录放的是 RFM 打分和 KMeans 分群这类算法代码，用 Scala 写而不是 SQL，因为涉及迭代计算和向量化，SQL 表达起来很勉强。bin目录里的 shell 脚本是最容易被新手忽略的部分，但它决定了任务能不能稳定跑在 YARN 上。

4.2 核心输出表：用户标签宽表、分群结果表与快照表

user_tag_wide是全项目的枢纽，下游模型和报表都从它取数。ads_user_segmentation是分群结果，用seg_date分区，方便回溯某一天的分群情况。快照表是维持数据可信度的关键：每天全量保存一份分群结果，保留最近 30 天，这样计算分群迁移率时，可以直接对比两周前的快照，而不需要重跑历史任务。

4.3 model 段的 RFM 打分代码：阈值这么设，结果才可信

RFM 是用户画像里最经典也最容易写错的算法。R 是最近购买时间，F 是购买频次，M 是消费金额。代码本身很简单，难的是阈值设定：

import org.apache.spark.sql.functions._ // 先算连续分，再拼成 RFM 码 val scored = baseDF .withColumn("r_score", when(col("recency_days") <= 7, 4) .when(col("recency_days") <= 30, 3) .when(col("recency_days") <= 90, 2) .when(col("recency_days") <= 180, 1) .otherwise(0)) .withColumn("f_score", when(col("frequency_cnt") >= 10, 4) .when(col("frequency_cnt") >= 5, 3) .when(col("frequency_cnt") >= 2, 2) .otherwise(1)) .withColumn("m_score", when(col("monetary_amt") >= 1000, 4) .when(col("monetary_amt") >= 500, 3) .when(col("monetary_amt") >= 100, 2) .otherwise(1)) .withColumn("rfm_code", concat(col("r_score"), col("f_score"), col("m_score"))) .withColumn("segment_score", col("r_score").cast("int") * 100 + col("f_score").cast("int") * 10 + col("m_score").cast("int"))

这里的阈值不是拍脑袋拍出来的。常见做法是先对三个字段分别跑一次分位数，看 P25、P50、P75 在哪里，再结合业务经验微调。比如客单价高的类目，1000 元可能只是最低档，而社区团购里 100 元已经是大额消费。concat生成四位的组合码只是中间产物，真正分群时会用segment_score做数值比较，避免字符串排序把“444”和“099”搞混。打分完成后记得统计每个分数段的用户量，如果某个分数段占比超过 50%，说明阈值分档不合理，需要重新切分。

4.4 spark-submit 提交与资源估算

集群搭建完成后，第一步不是调算法，而是摸清 YARN 队列的资源上限，再按数据量估算执行器配置。

spark-submit \ --master yarn \ --deploy-mode cluster \ --name user-profile-weekly \ --num-executors 10 \ --executor-cores 4 \ --executor-memory 8g \ --driver-memory 4g \ --conf spark.sql.shuffle.partitions=200 \ --class com.example.profile.UserProfilePipeline \ user-profile-1.0.0.jar --env prod --date 2025-06-16

num-executors、executor-cores、executor-memory三个参数要一起看，不能单独调。10 个执行器、每个 4 核 8G，意味着总并发度是 40，适合百 G 级别、有 5 到 10 个 SQL 阶段的画像任务。spark.sql.shuffle.partitions的经验值是executors × cores × 3 ~ 5，这里算下来是 120 到 200，取 200 留出余量。分区数太小会导致单个 task 处理数据量过大，分区数太大则 task 调度开销盖过计算本身——这个问题用 Spark UI 看 Shuffle 阶段的任务平均耗时就能判断。--date是业务日期参数，调度平台每天传当天日期，补数时只需要换日期重跑，不用改代码。

5. 数据挖掘高频踩坑记录：数据倾斜、脏 JSON 与标签漂移的排查

做画像项目最容易翻车的不是算法选型，而是四个隐蔽问题：热点 key 导致数据倾斜、JSON 解析偶发失败、标签口径悄悄漂移、shuffle 落盘涨到集群扛不住。前两个会让任务直接失败，后两个会让数据看起来正常但实际不可信。按排查顺序逐个讲。

5.1 现象一：某个 user_id 的 reduce 任务卡死，其他节点空等

现象：Spark UI 里某个 stage 的所有 task 都跑完了，只有一个 task 长时间运行，运行时长比其他 task 多一个数量级。

原因：数据倾斜。电商场景里最常见的倾斜 key 是“默认游客 ID”或者大促期间某个热门账号，它的行为日志量可能有几千万条，哈希分桶时全部落在同一个 task 上。处理手法是加盐拆分：

// 加盐：热点 key 先按盐值拆成多份，局部聚合后再合并 val saltedDF = orderDF .withColumn("salt", when(col("user_id") === hotUserId, (rand() * 10).cast("int")).otherwise(0)) .groupBy("user_id", "salt") .agg(sum("paid_amount").as("amt_part")) val totalDF = saltedDF .groupBy("user_id") .agg(sum("amt_part").as("paid_amount"))

先对热点用户单独加 0 到 9 的盐，把同一个用户的订单拆到 10 个 task 上做局部聚合，再去掉盐做全局聚合，倾斜就缓解了。注意盐的粒度要足够小，否则全局聚合时又会倾斜回去。更省事的做法是开启动态分区裁剪和 Adaptive Query Execution，让 Spark 自动处理部分倾斜，但热点用户非常突出时，还是手动加盐最稳。

5.2 现象二：JSON 文件里混入脏数据，Spark SQL 解析直接失败

现象：任务跑到行为日志解析阶段整批报错，错误信息指向 JSON parse failure，但看样本数据又觉得一切正常。

原因：spark.read.json默认做 schema 推断，一旦某个字段在大部分行里是字符串、少数行里是嵌套对象，推断结果就会不稳定。更常见的是某个日志文件里有几行非 JSON 文本，比如 L4 网关的半截日志。解决方法是显式声明 schema，并打开坏记录路径：

val rawDF = spark.read .schema(behaviorSchema) .option("badRecordsPath", "/data/bad/behavior_logs") .json(rawPath)

schema声明后，Spark 不再做全量推断，读文件速度快很多。badRecordsPath会把解析失败的行单独写到指定目录，而不是直接让任务失败。大促期间日志格式遇到上线变更，可以先看坏记录目录里的内容，确认是格式调整还是单纯脏数据，再决定改 schema 还是过滤。

5.3 现象三：同一标签昨天和今天算出来的数对不上

现象：运营反馈“上周的高价值用户数”和今天重算上周数据的结果差了 20%，代码没动，上游表也没动。

原因：标签口径漂移。最常见的是有人在优化 SQL 时改了一个 filter 条件，比如把pay_time从“时间戳字段”换成了“分区字段”，看起来对但统计的范围变了。另一个常见原因是上游某个源表做了补数，历史分区的数据变了，下游画像跟着变。解决办法是给每个标签加口径版本号，并在 SQL 文件头写清楚定义：

-- 标签口径：rfm_r_score -- 定义：最近购买时间距今 <= 7天 为 4 分，<=30天 为 3 分，<=90天 为 2 分，<=180天 为 1 分，否则 0 分 -- 变更记录：2025-06-01 修改阈值，30天档从 45 天调整为 30 天

同时建议在调度链路里加一个上游行数校验：对近 7 天每个分区的行数做波动检测，某天行数环比波动超过 30% 就报警。宁可先停任务也不要带着脏数据往下游跑。

5.4 现象四：shuffle 落盘暴涨，executor 频繁 GC 掉出集群

现象：任务能跑完，但 YARN 上看到的 Shuffle 写数据量是输入数据量的 5 倍以上，executor 日志里频繁出现 Full GC，甚至有 executor 失联导致任务重试。

原因：聚合 key 粒度过细，或者动态分区数设置过大。画像宽表组装时如果先在user_id和page_id两层做聚合，shuffle 中间量会爆炸。解决思路是先做粗粒度聚合，再在内存里做细粒度合并，同时打开 Spark 3.x 的自动分区合并：

spark-submit \ --conf spark.sql.adaptive.enabled=true \ --conf spark.sql.adaptive.coalescePartitions.enabled=true \ --conf spark.sql.adaptive.coalescePartitions.minPartitionsNum=40 \ --conf spark.sql.adaptive.shuffle.maxTargetPostShuffleInputSize=64m

参数含义是按照目标 shuffle 分区输入量自动合并小分区，减少 task 数量。minPartitionsNum=40限制了合并后分区数不能低于 40，避免合并过度导致单 task 内存溢出。如果开了 AQE 仍然频繁 GC，就要回到 SQL 本身看是否存在笛卡尔积式的 join，画像宽表组装里这类问题比想象中常见。

6. 画像上线前我只信这三道验证：分群效果、调度依赖与人工抽检

RFM 分数算出来、分群表也写完了，这时候最忌讳直接交给运营。我会先做三道验证，每一道都能拦住一类问题。第一道是分群稳定性验证，看连续两周的分群结果有没有出现离谱迁移。用快照表直接对比：

-- 连续两周分群迁移率：如果某群体留存率低于80%，先查口径变动 select a.segment_id, count(distinct a.user_id) as user_cnt, round(sum(if(b.segment_id = a.segment_id, 1, 0)) / count(distinct a.user_id), 4) as stay_rate from ads_user_segmentation a left join ads_user_segmentation_prev b on a.user_id = b.user_id where a.seg_date = '2025-06-16' group by a.segment_id

迁移率超过 20% 的群体，必须逐个看原因。第二道是调度依赖验证，重点检查上游任务是否真的跑完、分区是否就绪，以及重跑时能不能用一个日期参数完整回溯。第三道是人工抽检：随机抽 50 个用户，拉出他们最近三个月的订单和行为日志，照着宽表里的标签定义逐条核对，这个动作虽然原始，但能发现所有自动化校验发现不了的逻辑错误。

这套源码本质上是一个完整的 Spark 数据分析案例，值得投入的方向是把它改造成可配置的画像平台，而不是每次需求都重写一遍 SQL。我现在的习惯是每个画像项目都保留一份口径变更记录，哪怕只是在 SQL 文件头加一行注释，也能帮三个月后的自己省下半天排查时间。希望帮到你。

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