简介基于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原样落库的原始日志与业务快照用户行为日志、订单快照小时 / 天DWD清洗、去重、标准化后的明细行为明细、订单明细小时 / 天TAG 层按用户 ID 汇总的标签明细用户标签宽表天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 数据源盘点会员、订单、日志、商品四类数据怎么接一个典型的电商用户画像项目数据源大概有四类。数据源通常存储位置Spark 读取方式会员资料MySQL / 云数仓JDBC 批量拉取订单与售后Hive 数仓表Spark SQL 直接读用户行为日志Kafka 落 HDFS 的 JSONStructured Streaming 或批量 read.json商品与类目MySQL / 数仓维度表JDBC 或广播变量接入顺序有讲究。会员资料和商品维度表是小表直接全量拉不要做增量量不大而且变更频率低。订单表是大表按分区读取每天一个分区按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.xmlSQL 目录按分层命名文件名带序号表示执行顺序这一点很重要。etl目录只有一个 Scala 入口类负责串联流程启动 SparkSession、按依赖顺序执行 SQL、最后把宽表写入目标分区。model目录放的是 RFM 打分和 KMeans 分群这类算法代码用 Scala 写而不是 SQL因为涉及迭代计算和向量化SQL 表达起来很勉强。bin目录里的 shell 脚本是最容易被新手忽略的部分但它决定了任务能不能稳定跑在 YARN 上。4.2 核心输出表用户标签宽表、分群结果表与快照表表名粒度关键字段更新策略user_tag_wide每用户一行user_id, rfm_code, category_pref, active_level, ds每日覆盖ads_user_segmentation每用户一行user_id, segment_id, score, seg_date每日覆盖dim_user_id_mapping每个来源ID一行id_value, id_type, user_id, first_seen_dt每日增量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.partitions200 \ --class com.example.profile.UserProfilePipeline \ user-profile-1.0.0.jar --env prod --date 2025-06-16num-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.enabledtrue \ --conf spark.sql.adaptive.coalescePartitions.enabledtrue \ --conf spark.sql.adaptive.coalescePartitions.minPartitionsNum40 \ --conf spark.sql.adaptive.shuffle.maxTargetPostShuffleInputSize64m参数含义是按照目标 shuffle 分区输入量自动合并小分区减少 task 数量。minPartitionsNum40限制了合并后分区数不能低于 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 文件头加一行注释也能帮三个月后的自己省下半天排查时间。希望帮到你。本文还有配套的精品资源点击获取