简介本资源是一套基于Spark构建的电商用户行为分析系统完整实现面向计算机专业本科生毕业设计、大数据课程实践及Spark初学者解决真实业务场景下海量用户行为数据的采集、清洗、分析与可视化问题。资源包共286个文件含40个核心Java源码如SessionAggrStat、MockData、JDBCHelper等、185个XML配置文件涵盖Maven依赖与Spring集成、47个zbak备份文件以及PNG图表、Properties配置、Markdown文档等总大小1.28MB结构清晰、模块划分明确开箱即用。已有55人学习下载适合从环境搭建到算法实现全流程跟进。用户可直接获得高分毕设级项目源码导师评审99分、配套技术文档、Spark MLlib协同过滤推荐实现、Spark Streaming实时处理逻辑及ECharts可视化方案覆盖用户画像、点击流分析、会话统计等核心功能是深入理解大数据分析工程落地的优质实践样本。 如果两周前有人拍着我的肩膀说你会为了一套Spark电商用户行为分析系统的源码和文档熬夜到凌晨两点我一定觉得对方疯了。但这事真就发生了——起因是一份超过三千万条的电商脱敏行为日志和一句“做个用户行为分析出来”的需求。折腾完这一趟我觉得有必要把这套东西里的关键决策、踩坑过程、代码套路、集群搭建顺序全部沉淀下来于是有了这篇博文。无论你是刚学完Spark想找实战项目练手还是准备数据岗面试需要能讲清楚的项目或者公司内部正准备搭一套轻量级行为分析这篇内容应该都能帮你少走很多弯路。1. 电商用户行为分析系统解决了什么问题很多初学者会把用户行为分析理解成“PV/UV统计”真上了线才发现业务方要的东西完全不是这么回事。我曾跟运营同学认认真真聊过一轮发现他们真正关心的问题无非这几个今天有多少人在逛、这些人逛到什么程度才下单、什么东西卖得好、用户买完之后还会不会回来。这些问题翻译成大数据的语言恰好对应活跃度分析、转化漏斗、TopN商品、留存分析四个模块。这套系统要解决的核心问题就是把用户在产品里留下的痕迹——浏览、加购、下单、支付——变成业务方能直接看懂的结论。范围界定也很重要。一开始有人提议把推荐算法、实时风控、用户画像全塞进来我坚决否了。一个两三个人维护的项目第一版做太多功能只会让每个模块都稀碎。最终我划定的边界是离线批处理为主覆盖从原始日志到应用层指标的全链路输出结果落到MySQL和ES供一个简单报表页面查询。实时这块暂时不做留了接口以后扩展。系统内部按职责拆成了几块数据接入层负责读取HDFS上的原始日志ETL层负责清洗、格式化和标准化指标计算层是核心分模块计算活跃、漏斗、TopN和留存数据导出层把结果写到MySQL或者Redis最后是部署与调度层用crontab或Azkaban定时触发Spark作业。这个架构不复杂但胜在清晰每块都能单独测试、单独优化。复盘一下这个项目主要适合三类人。第一类是学完Spark基础但缺少完整项目经验的人可以通过它把RDD、Spark SQL、窗口函数、调优参数串起来第二类是准备面试的人这个项目可以作为简历上的独立项目能聊的点非常多从数仓分层到数据倾斜都能展开第三类是业务侧需要快速搭建一套成本可控的分析平台的小团队这套系统的思路和代码可以直接抄作业。信息量很大但我想先给个总体结论这套系统能做到什么问题输入是三千万条乱糟糟的行为日志输出是十几张业务能直接看的报表和图表单次全量计算在两台4核8G的服务器上跑完约二十分钟Spark技术栈下这个性价比已经很高了。2. 从埋点日志到分层数仓这套系统的数据底座设计2.1 一条行为日志是怎么设计的做用户行为分析第一个拦路虎是数据格式。当时我拿到的原始日志是JSON格式字段非常杂光时间就有三个版本有的带毫秒有的是纯日期字符串还有的时区都对不上。我重新设计了一套标准化的行为日志格式在ETL层统一处理。一条有效的用户行为日志至少应包含以下字段字段名类型含义说明userIdString用户ID脱敏后的唯一标识sessionIdString会话ID同一用户一次会话内相同eventTypeString事件类型view/cart/order/payitemIdString商品IDcategoryIdString商品类目IDprovinceIdString用户所在省份IDdeviceTypeString设备类型ios/android/pc/h5timestampLong事件发生时间的Unix时间戳毫秒级extraJSON扩展字段存放价格、数量、优惠券等这个设计的核心思路是“够用就好”。很多团队会一上来设计几十个字段结果一半以上日志里永远都是空值。我这套字段覆盖了行为分析里最常用的维度userId解决“谁”eventType解决“干了什么”timestamp解决“什么时候”itemId和categoryId解决“对什么感兴趣”deviceType和provinceId解决“从哪来”够用了。数据产生方面因为拿到的日志涉及用户隐私无法直接对外展示或分享源码我专门写了一个模拟数据生成器。它的逻辑不复杂按照泊松分布模拟用户活跃时间段比如晚8点到11点流量波峰按照电商类目比例分配商品浏览量以行业平均转化率约3%的比率给部分用户追加加购、下单、支付事件。用这个生成器可以批量产出千万级的行为日志写入HDFS。生成器本身也是项目源码的一部分方便任何人复现整个分析流程而不用等真实的业务数据。2.2 四层数仓建模ODS、DWD、DWS、ADS数据建模这块我直接照搬了大厂业界标准的数仓分层思路当然做了很大程度的裁剪。没有搞复杂到让人头晕的几百张表只保留了四层ODS层存储原始日志原始JSON文件落HDFS按天分目录目录格式/data/ods/behavior_log/20250101/。这一层有个原则就是“原封不动”即使日志里有脏数据也先保留不在这一层做任何加工因为一旦清理逻辑写错原始数据丢了就再也找不回来了。DWD层做清洗与明细标准化。把JSON拆成列格式化时间戳为可读的时间字段过滤掉userId为空、事件类型不在枚举范围内、时间戳超出合理范围的垃圾数据。这样处理后得到的明细宽表是后续所有指标计算的唯一数据源。DWS层做轻度汇总。针对高频查询场景预先按“用户天”粒度聚合出行为汇总表比如每个用户每天浏览了多少商品、加购了几次、下单了几次。这样ADS层计算留存、活跃的时候不需要再扫全表明细大大缩短分析链条。ADS层是应用层直接对接业务指标。每一张结果表对应一个主题比如用户活跃表、漏斗转化表、TopN商品表、留存表。这些表最终同步到MySQL供报表系统查询。最容易被忽略但也是实战中最重要的是DWD和DWS之间的“数据血缘”设计。我在每张表注释里都写明来源表和加工逻辑某一天指标对不上的时候顺着血缘能一路追查到原始日志。这一点在项目文档里占了很大篇幅很多人不写血缘出了问题才后悔。2.3 脏数据与边界情况的处理策略清洗过程中最耗精力的不是正常数据而是那些“看起来正常但算出来是错的”数据。我总结了三种高频脏数据处理方式如下。第一种是重复日志。同一个事件可能因为客户端重试被上报多次我用(userId, sessionId, eventType, itemId, timestamp)五个字段作为去重键在DWD层做dropDuplicates。这个键在正常情况下唯一重复上报的数据会被安全过滤。第二种是时间异常。日志里的时间戳集中在某几个前端服务器的本地时钟上有一批时间戳竟然比当前时间还晚一年。我加了时间范围过滤只保留“业务上线时间到当天1天”之间的记录超出的一律丢弃并单独统计丢弃率做监控。第三种是会话切割。sessionId是前端生成的但用户长时间停留后会话会超时如果再点会产生新的sessionId。我在计算会话相关指标时会在DWD层根据时间间隔重新切分同一userId相邻两条日志间隔超过30分钟就视为一次新会话。这个逻辑写起来很简单lag(timestamp).over(Window.partitionBy(userId).orderBy(timestamp))判断差值即可。这些边界处理看起来不起眼但它们决定了计算出来的指标是“业务敢拍板”还是“只敢自己看看”。我把每种处理规则都写进了项目文档的数据规范章节并配了对应SQL示例团队协作的时候特别有用。3. 技术选型复盘Spark凭什么而不是MapReduce或Flink这套系统选择Spark不是因为它叫Spark所以选了它而是我把MapReduce、Flink、Spark三者放在一起对比后做的决策。很多新手问“Spark和MapReduce有什么区别”“为什么不用Flink”这里把我的思考过程完整讲一遍。3.1 为什么不是MapReduceMapReduce在处理纯粹的单词计数这类任务时没有太大问题但用户行为分析的核心是“多次迭代计算”清洗之后要按用户聚合聚合结果要再按商品聚合中间还会涉及多个数据集的Join。MapReduce每个阶段的输出都要落一次磁盘下一个阶段再读回来这种序列化和磁盘IO的开销在复杂分析链路中会被无限放大。一个漏斗分析任务MapReduce可能要跑七八轮Spark只需要一轮DAG调度就完成了。在性能测试里同一个漏斗任务Spark比MapReduce快了将近六倍这个差距足够让我选型时直接排除MapReduce。可能有人会说可以用Hive来写啊底层Hive在Hadoop 2.x之后也支持Tez和MR多种执行引擎。但实际上Hive默认执行引擎还是MapReduce的时候同样面临这个问题而且Spark SQL可以直接复用Hive的元数据和SQL逻辑迁移成本非常低。在我这边最终方案是Spark SQL跑主要指标底层物理执行由Spark引擎完成等于把Hive的便利性和Spark的计算性能都占了。3.2 为什么不是FlinkFlink是流式计算领域的标杆如果你的场景里“实时”是刚需比如大促大屏要秒级更新或者要实时拦截恶意刷单那Flink确实更合适。但电商用户行为分析的常规需求是日报、周报、月度运营复盘数据次日产出完全够用。一旦选择Flink要面对的是一整套云原生部署、状态后端、checkpoint、事件时间语义的学习曲线对于小团队来说成本非常高。我做技术选型的原则是离线场景优先用Spark实时场景才考虑Flink。这套系统的定位是离线为主未来若要加实时看板可以单独起一个Flink任务处理同样的数据源输出结果写到同一张宽表供实时大屏查询跟离线链路天然互补不存在冲突。实际上Spark自身的Structured Streaming也可以处理部分准实时需求只是精力有限第一版没做。3.3 Spark生态内怎么选Spark SQL、RDD还是Streaming确定用Spark之后又要面对下一个问题同一个Spark框架里用RDD还是Spark SQL还是DataFrame我的经验是能写Spark SQL的尽量写Spark SQL因为它的Catalyst优化器会帮你做谓词下推、列裁剪、常量折叠这些自动优化而且代码可读性远高于一堆算子链。数据分析场景里SQL的表达力已经覆盖了90%的需求。RDD也不是完全不用。在ETL阶段需要处理特别复杂的自定义逻辑时比如某些非结构化的日志解析、复杂的session切分直接用RDD算子更灵活。我的最终分工是数据清洗用DataFrame API Spark SQL为主涉及面向对象封装的自定义处理用RDD算子两者之间通过df.rdd或rdd.toDF()互相转换。Spark Streaming暂时只在环境验证时跑了个wordcount确认环境没问题即可。3.4 Spark执行流程从提交作业到看到结果这个项目也是理解Spark执行流程的好素材。一次Spark作业提交后Driver端会启动SparkContext将作业拆分成多个StageStage之间按照宽依赖shuffle划分。每个Stage内部又拆成多个TaskTask被分发到Executor的线程池中执行。以漏斗分析作业为例第一步Driver拿到输入文件的元数据决定分区数生成一个初始RDD分区然后根据groupBy操作划出一个shuffle StageStage内部Task按key分区哈希分区把同一个userId的数据拉到同一个Executor上聚合。只有真正理解了这套“分区——Stage——Task”的模型你在写代码时才能预判哪些操作会触发shuffle哪些操作会把数据全部拉到Driver导致内存爆炸。举例来说有一次我图省事在collect()之前忘了查看数据量结果一个超大DataFrame直接拉回DriverJVM当场OOM。理解了执行流程之后就知道collect()这种行动算子是把所有分区的数据全部通过网络传输到Driver适合小结果集大结果集一定要用saveAsTable或write.format落地而不是collect。这个坑我详细记录在第7章调优部分。4. 四大核心分析模块的原理与实现路由下面进入整个系统最核心的部分——四个分析模块具体怎么算。每个模块我都会从业务口径、实现思路、关键SQL/代码三个层面说明这是源码里最值得读的部分也是面试中最容易被追问的部分。4.1 用户活跃度分析DAU、WAU、MAU怎么算才准活跃用户数的定义看似简单实际有巨大的口径差异。按“当天有行为”算还是按“当天有独立会话”算结果能差出20%。我最终采纳的口径是当天产生至少一条有效行为日志的用户去重数即为DAU周活跃是当周内有过任意行为的用户数月活跃同理。周和月都按自然周、自然月切分不采用滚动窗口。这里有个细节值得注意跨天会话怎么算。比如一个用户晚上11点50分开始浏览凌晨0点20分才下单如果按天切分这个会话会被拆成两天。处理方式是在DWD层把跨天会话记录延续到次日即当日活跃判断时把会话开始时间落在前一日但会话内存在当日行为事件的用户也计入当日DAU。SQL实现思路是先按sessionId聚合出会话开始时间和结束时间再与每日行为表join取交集。DAU的计算在Spark里最忌讳的是把全量明细拉到Driver再distinct。正确做法是SELECT date(event_time) as dt, COUNT(DISTINCT userId) AS dau FROM dwd_user_behavior WHERE dt 2025-01-01 GROUP BY date(event_time)当数据量大到COUNT(DISTINCT)都会拖慢作业时可以换成approx_count_distinct它基于HyperLogLog实现误差在2%以内但性能提升非常明显。我测试过三千万条数据approx_count_distinct比count(distinct)快了大约4倍对运营日报来说精度完全够用。4.2 转化漏斗从浏览到支付每一步流失在哪漏斗分析是整个系统里业务价值最高的模块。运营同学拿到漏斗图一眼就能看出哪个环节流失严重。我做的漏斗是“浏览→加购→下单→支付”四步结果表结构是步骤用户数相对上一步转化率相对第一步转化率view1200000-100%cart68000056.7%56.7%order24000035.3%20.0%pay19800082.5%16.5%实现思路是对每一步分别求“发生该行为的用户集合”相邻集合取交集人数除以父集合人数得到转化率。但要注意漏斗口径有两个版本。一个是会话级漏斗要求同一sessionId内完成浏览到支付才计入一个是用户级漏斗只要同一用户在某天内完成链路就算。会话级漏斗更能反映单次访问的转化效率用户级漏斗反映整体运营效果。我在系统里实现了两套通过参数切换默认输出用户级漏斗因为它的结果更稳定不容易被session切割逻辑干扰。关键代码用Spark SQL实现如下-- 第一步每个用户是否有过浏览行为 CREATE OR REPLACE TEMP VIEW step_view AS SELECT DISTINCT userId FROM dwd_user_behavior WHERE eventType view AND dt 2025-01-01; -- 第二步每个用户是否有过加购行为 CREATE OR REPLACE TEMP VIEW step_cart AS SELECT DISTINCT userId FROM dwd_user_behavior WHERE eventType cart AND dt 2025-01-01; -- 计算浏览→加购转化 SELECT COUNT(DISTINCT sv.userId) AS view_users, COUNT(DISTINCT sc.userId) AS cart_users FROM step_view sv LEFT JOIN step_cart sc ON sv.userId sc.userId;这里有个性能技巧千万级数据按eventType过滤后每个步骤的用户集合通常只有几十万到几百万后续的join实际量级不大。所以尽量先把过滤做在前面用“早过滤、晚聚合”的策略避免无效数据在全链路中反复流转。4.3 TopN商品与类目分析窗口函数的妙用TopN分析回答的是“什么卖得好”。细分为两个维度按商品维度找出指定时间窗口内浏览量最高的10个商品按类目维度找出销售额最高的TopN类目。这里的“销售额”不是订单金额全量求和而是支付事件里的实付金额需要在生成订单事件时明确带上我是把实付金额放进了extra字段解析时提取。实现TopN最优雅的方式是Spark SQL窗口函数WITH item_stats AS ( SELECT itemId, categoryId, COUNT(*) AS view_cnt, SUM(CASE WHEN eventType order THEN 1 ELSE 0 END) AS order_cnt FROM dwd_user_behavior WHERE dt 2025-01-01 AND dt 2025-01-07 GROUP BY itemId, categoryId ) SELECT itemId, categoryId, view_cnt, order_cnt FROM ( SELECT *, ROW_NUMBER() OVER (PARTITION BY categoryId ORDER BY view_cnt DESC) AS rn FROM item_stats ) t WHERE rn 10ROW_NUMBER() OVER (PARTITION BY categoryId ORDER BY view_cnt DESC)这个窗口的意思是在每个类目分区内按浏览量降序编号取编号1到10就是每个类目下浏览量Top10的商品。初学阶段很容易绕晕一个直观类比是分组考试排名PARTITION BY是分考场ORDER BY是看谁分数高ROW_NUMBER是每个考场内的排名编号。这个模块最容易被忽略的是“重复请求清洗”。有些爬虫或异常脚本会产生极高的浏览量导致TopN永远被同一批非人类行为霸榜。我在DWD层做了一次基础过滤同一userId对同一itemId的浏览事件5分钟窗口内只保留一条。这样能让排名更接近真实用户偏好也让结果对运营有参考价值。4.4 用户留存分析次日、7日、30日应该怎么定义留存分析的难点在于“第一天的基准是什么”。我采用的基准是用户在统计周期内的首次活跃日作为“激活日”然后看该用户在激活日后的第N天是否产生了行为。如果第N天有行为记为留存否则记为流失。这种口径叫“首次激活留存”比“任意活跃用户留存”更能反映产品质量的变化。计算留存的关键是一张“用户首次活跃日”注册表。伪代码如下-- 首次活跃日 CREATE OR REPLACE TEMP VIEW user_first_active AS SELECT userId, MIN(date(event_time)) AS first_active_date FROM dwd_user_behavior GROUP BY userId; -- 次日留存 SELECT f.first_active_date, COUNT(DISTINCT f.userId) AS new_users, COUNT(DISTINCT IF(DATE_ADD(f.first_active_date, 1) b.event_date, b.userId, NULL)) AS retain_1d_users FROM user_first_active f LEFT JOIN ( SELECT userId, date(event_time) AS event_date FROM dwd_user_behavior WHERE dt 2025-01-01 AND dt 2025-02-01 ) b ON f.userId b.userId GROUP BY f.first_active_date这里有几个坑要先说明。第一个坑是“留存用户数必须小于等于新增用户数”如果出现某一列大于基准列基本都是join时把一对多数据拆开了导致重复计数需要用COUNT(DISTINCT ...)来规避。第二个坑是日期偏移要确认Spark SQL的DATE_ADD函数是否按你的业务日切分如果你用的是UTC时间而业务在东八区需要先做时间偏移。第三个坑是历史拉链问题留存计算会随着时间推移每天都重算要确保只更新最近30天的分区否则全量重算会非常浪费资源。留存分析的结果会写入ADS层留存表以用户首次活跃日为主键每日更新各N日留存率。运营可以按月拉出“这批新用户第30天还剩多少”用来评估活动拉新质量。5. 集群环境从零搭建资源规划、部署顺序与验证方法源码写好只是第一步环境搭不起来后面全是空谈。这章按照我的实际操作顺序来写每一步都标注为什么这样做。5.1 硬件资源怎么规划很多人一开始就纠结需要多大集群我的建议是前期学习和验证阶段用两台服务器就够了资源不必贪多。我用的配置是两台4核8G内存的云主机磁盘各100G跑三千万条日志够用。真实生产环境起步可以按“数据量每日新增1GB以内离线日批”这个量级来算一般三台节点、每台8核16G是中位数配置。内存规划有个经验公式Spark作业的总内存需求约等于“最大输入数据集的2倍到3倍”。比如日增量1GB处理时会有临时shuffle数据需要的内存至少2GB到3GB加上Executor JVM开销一台8G内存的节点实际能用于Spark的内存也就4到5G。我当时就是按这个估算选了8G内存的节点跑起来分区数调到了24整体刚好处于稳定区间。如果数据量翻倍优先增加Executor数量或者每个Executor的内存而不是盲目提高分区数。分区数过大会导致线程切换开销过小则单个任务处理数据过多容易OOM需要找到一个平衡点。5.2 前置依赖JDK、Hadoop、Spark的版本匹配版本匹配是环境搭建的第一大坑。Spark 3.x要求JDK8或JDK11Hadoop的版本必须在Spark预编译的对应范围内否则提交作业时会报Unsupported class file major version或者HDFS客户端协议不兼容之类的错误。我这里选的是Spark 3.3.0 on Hadoop 3.3.2JDK用了1.8。选择这个组合的原因是社区验证充分、网上资料多出问题好查。下载和解压Spark之后最关键的配置文件是spark-env.sh要设置好JAVA_HOME、HADOOP_CONF_DIR和SPARK_MASTER_HOSTexport JAVA_HOME/usr/local/jdk1.8.0_202 export HADOOP_CONF_DIR/usr/local/hadoop/etc/hadoop export SPARK_MASTER_HOSTnode01 export SPARK_WORKER_CORES4 export SPARK_WORKER_MEMORY4g如果单机学习可以不开YARN用Spark standalone模式即可。standalone模式的好处是部署简单、Web UI直观适合跑通流程。生产环境则建议上YARN让资源统一调度多个作业之间不会互相抢占。我用的是YARN模式所以在spark-submit时指定--master yarn。5.3 部署顺序HDFS、Spark、MySQL部署顺序通常是先装HDFS再装Spark最后装MySQL结果存储。HDFS启动后先建好根目录比如/data/ods和/data/dws再把模拟生成的日志文件用hdfs dfs -put上传到ODS目录。上传完成后在Spark里写一个简单作业读取JSON并打印schema验证链路通断。验证环境的最短命令是跑一次spark-shellbin/spark-shell --master yarn --deploy-mode client进入shell之后执行spark.read.json(hdfs://node01:9000/data/ods/behavior_log/20250101/).printSchema()能正常打印出字段结构说明HDFS、YARN、Spark三者之间的链路已经通了一半。接着再跑一个groupBy(eventType).count()如果结果秒出环境基本就没问题了。MySQL这边主要是建库建表。我设计的结果表按天分区使用REPLACE INTO写入保证重复跑同一批数据不会产生脏数据。写入MySQL时要注意驱动jar包的位置把它放到$SPARK_HOME/jars目录下否则会报ClassNotFound。这个坑我印象太深了第一次跑导出脚本时卡了整整一下午结果发现就是驱动没放进jars目录。5.4 提交作业的参数配置一份可以直接抄的模板环境准备好后跑的作业参数直接决定了作业能不能稳定运行。我调好的参数模板如下spark-submit \ --master yarn \ --deploy-mode client \ --driver-memory 2g \ --num-executors 4 \ --executor-memory 3g \ --executor-cores 2 \ --conf spark.sql.shuffle.partitions24 \ --conf spark.memory.fraction0.6 \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ --class com.example.UserBehaviorAnalysis \ user-behavior-analysis.jar说明一下参数含义--executor-memory是每个Executor的JVM总内存--executor-cores是每个Executor并发执行的Task数--num-executors是Executor个数三者乘积约等于集群可并行度spark.sql.shuffle.partitions控制shuffle后结果的分区数默认200对于小数据集来说太多会导致大量空Task和网络开销我调成24spark.memory.fraction决定统一内存中Spark可用比例默认0.6如果数据有大量缓存需求可以调到0.7以上KryoSerializer相比Java序列化能大幅减小序列化体积代价是需要注册类对于包含复杂对象的任务收益明显。6. 核心代码设计ETL、算子链与Spark SQL窗口函数实战源码组织得好不好直接决定了项目能不能被别人看懂。6.1 工程结构按职责分层而不是按功能平铺我最终的工程结构是Maven管理的Java/Scala混合工程但核心业务逻辑全部用Scala实现。结构如下user-behavior-analysis/ ├── pom.xml ├── README.md ├── docs/ │ ├── 01-架构设计.md │ ├── 02-数据规范.md │ ├── 03-部署指南.md │ ├── 04-调优手册.md │ └── 05-FAQ.md ├── src/ │ ├── main/ │ │ ├── java/ # JDBC工具类等少量Java类 │ │ └── scala/ │ │ ├── common/ # 常量、参数解析、SparkSession工厂 │ │ ├── etl/ # 清洗作业 │ │ ├── metrics/ # 活跃、漏斗、TopN、留存四个指标作业 │ │ └── export/ # MySQL导出作业 │ └── test/scala/ # 单元测试与本地验证脚本 ├── bin/ │ ├── run_etl.sh │ ├── run_metrics.sh │ └── run_export.sh ├── sql/ │ ├── create_tables.sql │ └── sample_queries.sql └── data/ └── mock_generator/ # 模拟数据生成器每个指标作业的入口都遵循同一个逻辑模板解析参数、初始化SparkSession、调用对应计算函数、写结果表。这样别人拿到源码后看任一个指标作业的代码其余三个的套路都能猜到一半。用统一的代码结构来降低项目理解成本比什么都重要。6.2 SparkSession的初始化与参数解析SparkSession是Spark 2.0之后统一的入口初始化时要把前面调优的参数固化在代码里而不是散落在各个shell脚本中。我的SparkSessionFactory里核心一段object SparkSessionFactory { def create(appName: String, params: Map[String, String]): SparkSession { val builder SparkSession.builder() .appName(appName) .config(spark.sql.shuffle.partitions, params.getOrElse(shuffle.partitions, 24)) .config(spark.sql.crossJoin.enabled, true) .config(spark.sql.adaptive.enabled, true) .config(spark.serializer, org.apache.spark.serializer.KryoSerializer) if (params.getOrElse(master, yarn) local) builder.master(local[*]) builder.getOrCreate() } }spark.sql.adaptive.enabled是Spark 3.0引入的动态分区裁剪和合并机制开启后环境会根据运行时的数据量自动调整分区数这个默认开着的参数帮我减少了很多手动调分区的动作。如果你用的是Spark 2.x需要手动写更多调优逻辑所以我强烈建议新项目直接用Spark 3.x。6.3 ETL清洗作业从JSON到明细宽表ETL作业的整体链路是读取原始ODS JSON → 解析 → 过滤 → 标准化 → 写DWD层Parquet。Parquet列式存储比JSON节省大量存储空间而且Spark SQL读Parquet会自动做列裁剪和谓词下推极大提升后续指标计算的效率。核心清洗逻辑如下val rawDF spark.read.json(inputPath) val dwdDF rawDF .filter($userId.isNotNull $eventType.isNotNull) .filter($eventType.isin(view, cart, order, pay)) .filter($timestamp startTime $timestamp endTime) .withColumn(event_date, from_unixtime($timestamp / 1000, yyyy-MM-dd)) .withColumn(event_hour, hour(from_unixtime($timestamp / 1000))) .withColumn(extra_price, get_json_object($extra, $.price)) .dropDuplicates(userId, sessionId, eventType, itemId, timestamp)这里有个容易踩的坑from_unixtime($timestamp / 1000, yyyy-MM-dd)默认使用JVM时区如果在服务器上没设置好时区生成的event_date会差8个小时。我在所有节点配置了-Duser.timezoneGMT8并在SparkSession启动时加上spark.sql.session.timeZoneAsia/Shanghai确保时间字段口径统一。时间相关的口径错误是最隐蔽的不逐一对账很难发现。6.4 指标计算的代码实现以TopN为例TopN的完整代码段如下我直接在metrics包的TopNItemJob里实现了类目Top10商品object TopNItemJob { def main(args: Array[String]): Unit { val params ArgParser.parse(args) val spark SparkSessionFactory.create(TopNItemJob, params) val dwdPath params(dwd.path) val outputPath params(ads.path) /item_topn/ val dwdDF spark.read.parquet(dwdPath) .filter($dt params(dt)) val topN dwdDF .filter($eventType view) .groupBy(itemId, categoryId) .agg(count(*).alias(view_cnt)) .withColumn(rn, row_number().over( Window.partitionBy(categoryId).orderBy(desc(view_cnt)) )) .filter($rn 10) .select(itemId, categoryId, view_cnt, rn) topN.write.mode(overwrite).parquet(outputPath) spark.stop() } }这段代码里的Window.partitionBy(categoryId).orderBy(desc(view_cnt))非常典型。注意窗口函数在数据量极大时会产生单个分区的单点压力如果某个类目下的商品数量特别巨大比如“全品类”这种分类方式会导致某个分区处理数据量远大于其他分区。对这种场景可以先在内存中做一次粗筛比如先算出全量Top100再按类目做窗口既节省计算又避免单点。6.5 结果导出写入MySQL的正确姿势ADS层结果最终要同步到MySQL我写了一个通用的导出工具类核心代码如下val df spark.read.parquet(adsPath) df.write .mode(SaveMode.Append) .jdbc(jdbcUrl, tableName, connectionProps)这里要注意SaveMode.Append重复执行会产生重复数据。我的方案是在写入前执行一次DELETE FROM table WHERE dt targetDate然后再写入。而SaveMode.Overwrite对MySQL来说可能会删掉整张表再重建如果表里有历史数据就悲剧了。所以输出模块统一采用“先删分区-再Append”的策略这是我在生产环境被坑过一次后总结出来的。另外MySQL表主键设置为(dt, userId)这类自然键写库时使用REPLACE INTO幂等语义但Spark JDBC默认用INSERT需要配置rewriteBatchedStatementstrue和useServerPrepStmtstrue来优化大批量写入速度。这些配置在connectionProps里加就行。7. 性能调优实战数据倾斜、GC压力和小文件治理这一章是项目里最值钱的部分。源码跑通容易跑得快、跑得稳才是真功夫。7.1 数据倾斜某几个Task卡到天荒地老现象很典型提交一个聚合作业进度条跑到99%但剩下最后两三个Task死活跑不完日志里看到某个Task处理的数据量是其他Task的几十倍。这就是数据倾斜。定位方式首先要看Spark Web UI上的Stage详情页对比各个Task的Shuffle Read大小。正常情况下数据量应该大致均匀如果一个Task读到几个GB而其他Task只有几十MB基本可以锁定是数据倾斜。我的项目里出现这个问题的场景是按categoryId分组时某个“全品类”的categoryId下面挂了大量商品窗口函数对这个分区做排序时单个分区要处理的数据量远超其他分区。解决方案有几种按优先级我推荐这种排序两阶段聚合加盐先用rand对key加随机前缀打散做一次局部聚合然后去掉前缀再做全局聚合。适用于groupBy聚合后倾斜的场景。广播join小表通过spark.sql.autoBroadcastJoinThreshold默认10M阈值自动广播如果小表超过阈值可以手动broadcast(df)避免Shuffle。分区裁剪在join之前根据业务逻辑先缩小一边数据量比如只取最近7天数据再join能大幅减少倾斜程度。我的topN处理方式属于“预聚合小结果集”方案先全局算一个Top1000这个阶段数据量大但只是普通聚合不会产生单点倾斜然后基于这1000条再做精确窗口排序。这样即使某个类目商品特别多聚合阶段也是分散的不会集中在某个分区。7.2 内存瓶颈与GC压力从OOM到稳定运行的排查过程跑TopN商品排名时第一次全量提交直接报错Container killed by YARN for exceeding memory limits。这个报错最坑的一点是它经常出现在Executor内存已经爆了但YARN还没等到心跳响应的时候表面上看是“内存超限”实际上是GC长期停顿导致心跳超时。排查步骤我记录得很清楚先在Spark Web UI看Executor的GC Time列。正常值应低于总运行时间的10%如果超过了基本可以断定GC是主要瓶颈。看Storage Memory使用情况。如果缓存数据占了大量内存且任务本身不是迭代型任务就可以关掉缓存或者降低spark.memory.storageFraction。排查是否存在大量对象在shuffle后才释放。我的问题就是窗口函数的全局排序导致单个分区持有了大量对象GC频繁Full GC最终被YARN杀掉。解决方案是组合拳一是把--executor-memory从3g调到4g让JVM有更多堆内存二是把spark.memory.offHeap.enabledtrue分配1g的堆外内存给序列化缓冲三是优化了窗口排序的写法改成“预聚合后再排序”的方式让进入排序阶段的数据量下降80%。修改后作业运行时间从45分钟降到了20分钟。还有一次是Driver端OOM原因是某个指标计算后collect()了一张大表到Driver。这个问题的根因就是我在6.3提到的不规范开发习惯没有先估算结果集大小就直接collect。后来我规定所有指标作业里禁止collect()结果一律落HDFS或JDBCDriver侧内存负载大幅下降。7.3 小文件问题Spark写Parquet的“隐形杀手”Spark写Parquet时默认会按分区自动生成文件但如果分区数设置得太大每个Task写出来的文件很小比如只有几十KB最终结果就是HDFS上冒出几千个小文件。小文件会严重影响下游读取NameNode内存被大量占用读取时会频繁进行文件元数据请求整体速度反而下降。我在DWS层生成用户汇总表时就遇到了这个问题写出来的文件有七八百个每个不到几百KB。解决办法是写之前控制分区数让每个输出文件大致达到128MB左右dwdDF.repartition(col(event_date), col(provinceId)) .write .partitionBy(event_date) .option(maxRecordsPerFile, 500000) .parquet(outputPath)repartition(col(event_date), col(provinceId))把数据按天、省份打散避免某个分区过大maxRecordsPerFile限制每个文件最多50万条记录超了会自动拆文件。这样文件大小维持在几十MB到一两百MB之间既不会太大不好处理也不会太小产生海量元数据。如果已经产生大量小文件最直接的办法是重分区后覆盖写一遍或者用spark.sql.adaptive.coalescePartitions.enabledtrue在读取时自动合并小分区。Spark 3.x的自适应执行特性会一定程度上缓解这个问题但不能完全依赖它写入侧控制才是根本。8. 源码与文档怎么组织才能让别人真的看懂、跑得起来源码和文档是这套系统的对外交付物很多项目代码写得好但没人看得懂或者文档写得天花乱坠但照做就报错。我在这部分花了很多心思最终沉淀了一套比较实用组织方案。8.1 源码目录设计让人三分钟找到入口目录设计的原则是“入口要近、逻辑要清”。我前面展示过完整目录这里重点说明几个容易被忽略的细节。第一bin/目录下放了三个启动脚本分别对应ETL、指标计算、导出三步脚本里把所有参数都抽成了变量顶部有注释说明每个参数的用途。这样运维同学不用打开源码就能部署只需要改环境相关的配置。第二sql/create_tables.sql里包含了MySQL所有结果表的建表语句表名规范统一ads_user_active、ads_funnel、ads_item_topn、ads_user_retention一看就知道属于哪个主题。表设计上统一加了dt分区字段后续做增量更新、数据重算都非常方便。第三每个指标类的注释规范我强制自己在类头写上“输入表”“输出表”“计算口径”“注意事项”四段。举个例子/** * 漏斗分析作业 * 输入: dwd_user_behavior (按dt过滤) * 输出: ads_funnel (按dt分区) * 口径: 同一用户当天依次完成view/cart/order/pay四步 * 注意: 用户级漏斗跨天会话不合并 */写注释的初衷很简单三个月后的自己也是陌生人。如果连自己都看不懂自己写的代码那这套源码对别人来说就更难了。8.2 文档清单五份文档把项目讲透完整文档不是一篇长文而是按角色拆开的五份独立文档各自面向不同的使用场景。架构设计文档面向所有阅读者包含系统边界、模块划分、数据流图用文字描述版、技术选型理由、表结构说明。数据规范文档面向数据开发和运营定义每个字段的枚举值、时间口径、去重规则、注意边界。部署指南面向运维从JDK安装到Spark集群搭建到MySQL建库每一步都给出命令和验证方法。调优手册面向后续维护者详细记录我在第7章描述的三类性能问题和解决方案。FAQ文档则汇集了初跑项目时所有人都会遇到的十几个问题比如“找不到驱动类怎么办”“YARN队列资源不足怎么办”“时区差8小时怎么排查”。部署指南里最实用的是把“从零到全流程跑通”压缩成了六个步骤修改bin/run_etl.sh里的输入路径和日期参数执行bin/run_etl.sh跑通ETL修改bin/run_metrics.sh执行bin/run_metrics.sh生成四个指标结果执行bin/run_export.sh把结果写入MySQL打开MySQL查询ads_funnel表看数据是否正常用sql/sample_queries.sql里的样例查询验证结果量级是否合理。8.3 快速复现换一台机器怎么跑通我把这套项目部署到另一台全新服务器上花的时间从最初的完整一天缩短到了两小时主要靠的是部署文档里的一份“环境版本对照表”。所有组件的版本号都有精确对应关系照抄即可组件版本说明JDK1.8.0_202不要用OpenJDK 11以上的版本有兼容隐患Hadoop3.3.2对应Spark预编译的hadoop版本范围Spark3.3.0源码编译时用的版本Scala2.12.15Spark 3.3默认编译的Scala版本MySQL8.0存储ADS层结果然后执行模拟数据生成器data/mock_generator/下的脚本生成测试数据上传HDFS按部署文档的六步走基本一小时内能跑通全流程。我特意把模拟数据生成器做得非常轻量甚至可以在单机Spark local模式下跑这样即使没有Hadoop环境也能在IDE里把整条链路跑完。8.4 后续扩展方向实时化和图计算最后说下这套系统的扩展性。虽然第一版定位离线但代码结构上给实时化留了接口。未来如果要做实时大屏可以用Structured Streaming消费Kafka里的行为日志复用DWD层的口径计算逻辑直接输出到RedisApp或网页端读Redis就能展示。图计算方向可以算用户的社会关系网络比如同一设备下登录过多个账号或者多个用户共享同一收货地址这在风控场景很有价值。图计算用GraphX或GraphFrames实现但改动会比较大一般放到二期再做。我个人实际操作的体会是源码和文档的质量决定了一个项目能走多远代码是为了让机器读懂文档是为了让人在三个月后还能读懂两者缺一不可。很多团队项目代码写完就扔给运维最后运维看不懂开发改不动项目变成一坨屎山。这个教训我体会太深了。最后再分享一个小技巧。如果你准备把这个项目写到简历上千万别只写“基于Spark的电商用户行为分析系统”这种标题一定要把项目的核心指标、数据规模、性能表现写清楚比如“接入日均千万级行为日志依托Spark SQL完成四种核心分析通过数据倾斜和小文件治理使作业耗时降低50%”。面试官看到这种描述才知道你是真的做过、真的跑过、真的调优过而不是把网上的项目抄了一遍。本文还有配套的精品资源点击获取