简介面向大数据开发与Spark学习者的电商用户行为分析平台项目提供完整源码与模拟数据集。项目覆盖用户session分析、页面单跳转化率统计、热门商品离线统计、广告流量实时统计四大业务模块涉及Spark Core、Spark SQL、Spark Streaming三套技术框架并融入数据倾斜处理、线上故障排查、性能调优等进阶经验。压缩包共236个文件以40个Java源码、184个XML配置为主另含6张PNG示意图整体大小仅1.23MB目录结构清晰便于对照学习。整套资源从需求分析、方案设计到编码测试与调优均有体现模拟数据可直接运行验证已有387人学习下载适合具备Spark基础、希望提升大数据项目实战能力的读者。1. 用户行为分析不是写几个 SQL 就能交付的接手这套电商用户行为分析平台之前我一度以为把访问日志导进 Hive写几条聚合 SQL 就能出数。真正跑起来才发现业务方要的不是一张表而是一连串连环问题今天新增了多少有效 sessionsession 时长的分布曲线长什么样从首页到商品详情页的转化掉了多少哪些品类值得上首页推荐广告点击流量在分钟级有没有异常波动。这些需求横跨离线批处理和实时流处理只靠 SQL 或者只靠一个计算框架都撑不住。这套基于 Spark 的电商用户行为分析大数据平台源码把四个业务模块串成了完整闭环用户 session 分析、页面单跳转化率统计、热门商品离线统计、广告流量实时统计。工程里既有 Spark Core 的 RDD 算子实战也有 Spark SQL 的结构化查询还有 Spark Streaming 的实时计算以及贯穿全程的数据倾斜与性能调优经验。适合有两到三个月 Spark 基础、想完整拆一个企业级项目的开发者也适合正在设计数据平台、需要参考模块划分和技术选型的一线工程师。读源码的价值不只是跑通流程而是看作者在需求分析、数据设计、编码实现、测试和调优这几个环节里到底做了哪些取舍。2. 平台分层与 Spark 三个子框架的边界2.1 MockData 打点逻辑与数据模型平台跑在模拟数据上所以先要读懂MockData.java里数据是怎么造出来的。模拟数据覆盖用户访问行为、用户基本信息和商品信息三类表其中用户访问行为表是核心字段包含 date、user_id、session_id、page_id、action_time、search_keyword、click_category_id、click_product_id、order_category_ids、order_product_ids、pay_category_ids、pay_product_ids。一个 session_id 代表用户一次连续的访问会话行为类型通过字段是否有值来区分比如同时有 click_category_id 和 click_product_id 说明是一次点击行为order_category_ids 非空说明产生了下单pay_category_ids 非空说明完成了支付。模拟数据生成要注意两个参数数据量和 session 的分布密度。一般做法是控制 user_id 在某个区间内随机比如 1 到 100 的用户但把 session_id 的基数放大到 10000 以上这样 groupBy session 时才会产生足够多的分组聚合统计才有意义。另一个关键是行为时间要递增不能出现同一个 session 里时间倒挂否则后续 session 时长和单跳转化率的计算结果都是错的。造数时倾向于把 70% 的行为集中在少数热门 session 上剩下 30% 均匀分布刻意制造一些倾斜等第四模块调优时就能直接看到效果。2.2 源码工程里的核心对象与职责项目工程结构不复杂核心类之间的协作关系却很清晰我在下表中整理了主要对象的职责边界。源码对象职责关键方法/属性MockData.java生成模拟访问行为、用户信息、商品信息数据generateUserVisitActionData()UserVisitAnalyze.java用户 session 分析主入口负责按时间区间聚合 session过滤有效 sessionmain()、sessionAggrStat()SessionAggrStat.java对 session 的访问时长、访问步长做分段统计calculate()CategorySortKey.java实现品类排序的二次排序 Key比较点击量、下单量、支付量compareTo()JDBCHelper.javaMySQL 写入工具封装批量插入、连接复用、事务控制insertBatch()DateUtils.java日期格式化、时间差计算用于 session 时长和单跳时间间隔getTimeDifference()SessionDetail.java保存 session 明细数据供聚合后的明细查询使用actionList模块之间通过 RDD 和 DataFrame 传递数据JDBCHelper是唯一的数据出口所有统计结果最终都落到 MySQL。CategorySortKey这个类是重点它不是普通排序而是让 Spark 在 reduceByKey 或 sortByKey 时按点击优先、其次下单、最后支付的顺序排这是电商热门商品统计里非常典型的排序需求。2.3 Spark Core、SQL 与 Streaming 的分工不是拍脑袋读这个项目最大的收获之一是搞清楚为什么同一套平台里要同时用 Spark Core、Spark SQL 和 Spark Streaming而不是图省事全用 RDD 一把梭。用户 session 分析这一步逻辑复杂需要先按 session_id 分组再对每组的访问时长和步长做条件过滤和分段这种逐条数据的分支判断用 RDD 算子写起来最直接。Spark Core 的 map、filter、reduceByKey 在处理这种非结构化、带大量业务分支的逻辑时灵活性和可控性比 DataFrame 高尤其是调试阶段可以随时打印中间结果。页面单跳转化率和热门商品统计的数据形态是规整的日志表过滤、关联、聚合的语义非常明确用 Spark SQL 最划算。同样一段取每个页面 PV 的逻辑DataFrame 的 groupBy 和 agg 写出来不到十行换成 RDD 要处理各种序列化和类型转换。这里要注意一个容易踩的坑Spark SQL 的spark.sql.shuffle.partitions默认值是 200如果数据量只有几百 MB200 个分区反而会让每个 task 处理的数据量过小shuffle 文件过多调度开销变大。小数据集跑批任务时我一般把它调成数据量 / 128MB的倍数比如 1GB 的数据量设置为 8 到 16 个分区。Spark Streaming 的任务是广告流量实时统计它和离线任务的边界在于时间和状态。离线任务处理的是今天之前的所有数据实时任务处理的是最近几秒进来的数据。如果广告点击流也要做历史累计则必须在 Streaming 里维护跨批次的状态这就要用到 updateStateByKey 或 mapWithState。三个框架在项目里不是堆砌而是根据数据的时效性和计算逻辑的复杂性分工数据越规整、时效性要求越高就越往 SQL 和 Streaming 靠逻辑分支越多、数据越杂越需要 Core 的算子能力兜底。3. Session 聚合与页面单跳转化率的实现细节3.1 Session 切分逻辑与步长时长分段统计用户 session 分析的起点是从原始访问行为数据中切分出一个个 session。切分规则通常是同一个 session_id 内的行为时间间隔不超过 30 分钟超过 30 分钟视为另一次会话。工程里的实现方式是先读取原始访问行为数据过滤出指定时间范围内的记录然后按 session_id 做 groupByKey再对每组内部的 action_time 排序。排序这一步容易被忽略因为 groupByKey 之后 RDD 的分组内部并不保证有序如果直接遍历计算步长会出现时序错乱。JavaPairRDDString, IterableUserVisitAction sessionActions actionRDD.groupByKey(); JavaPairRDDString, SessionAggrStat sessionStatRDD sessionActions .mapToPair(tuple - { String sessionId tuple._1(); IterableUserVisitAction actions tuple._2(); ListUserVisitAction actionList new ArrayList(); for (UserVisitAction action : actions) { actionList.add(action); } // 按访问时间排序保证步长和时长的计算顺序正确 actionList.sort(Comparator.comparing(UserVisitAction::getActionTime)); long visitLength 0L; long stepLength 0L; long startTime 0L; long endTime 0L; for (int i 0; i actionList.size(); i) { UserVisitAction action actionList.get(i); if (i 0) { startTime DateUtils.parseToLong(action.getActionTime()); } if (i actionList.size() - 1) { endTime DateUtils.parseToLong(action.getActionTime()); } stepLength; } visitLength (endTime - startTime) / 1000; return new Tuple2(sessionId, new SessionAggrStat(sessionId, visitLength, stepLength)); });这段代码完成的是 session 步长和时长的初步计算。先对每个 session 内的行为按时间排序然后遍历一次拿到第一条和最后一条行为的时间戳算出访问时长同时累加行为条数作为访问步长。这里有个关键细节访问时长是按秒算还是按毫秒算直接决定了后面分段统计的边界值。项目里统一按秒计算因为业务方看的是分钟和小时的分布秒级精度足够。拿到每个 session 的时长和步长之后SessionAggrStat负责做分段统计。分段的方式常见的是区间映射访问时长分成 0 到 3 秒、4 到 10 秒、11 到 30 秒、31 到 60 秒、1 到 3 分钟、3 分钟以上这几档访问步长分成 1 到 3 步、4 到 6 步、7 到 9 步、10 到 30 步、30 步以上。实现时避免用 if-else 堆一串边界判断更优雅的方案是先把边界值放进一个数组遍历数组用二分查找定位区间下标再用一个累加数组计数。3.2 页面单跳转化率的计算链路页面单跳转化率是电商分析里最容易被业务方追问的指标之一。它衡量的是用户从页面 A 跳转到页面 B 的概率比如从首页跳到搜索结果页的转化率是多少从搜索结果页跳到商品详情页的转化率是多少。计算链路分两步第一步统计每个页面被访问的总次数PV第二步统计从页面 A 跳转到页面 B 的次数。-- 第一步统计每个页面的 PV SELECT page_id, COUNT(*) AS pv FROM user_visit_action WHERE date 2024-11-20 GROUP BY page_id;第二步的跳转次数统计不能直接用 SQL 的 GROUP BY 完成因为需要识别相邻的访问行为。工程里的实现方式是先按 session_id 分组在每个 session 内部把行为按时间排序然后生成相邻页面对。这里用到了 RDD 的mapPartitions在分区内处理排序后的列表遍历时每遇到相邻两次访问行为就输出一个以页面A_页面B为 key 的计数。JavaPairRDDString, Long pageJumpPairRDD sessionActions.flatMapToPair( (FlatMapFunctionTuple2String, IterableUserVisitAction, Tuple2String, Long) tuple - { ListUserVisitAction actions new ArrayList(); tuple._2().forEach(actions::add); actions.sort(Comparator.comparing(UserVisitAction::getActionTime)); ListTuple2String, Long list new ArrayList(); for (int i 0; i actions.size() - 1; i) { String fromPage String.valueOf(actions.get(i).getPageId()); String toPage String.valueOf(actions.get(i 1).getPageId()); list.add(new Tuple2(fromPage _ toPage, 1L)); } return list.iterator(); });生成跳转对之后用reduceByKey累加每个页面A_页面B的总次数。转化率的计算不能只看跳转次数必须拿跳转次数除以起始页面的总 PV这样才能消除页面本身流量大小的影响。还有一个工程上的迭代点页面集合是固定的很多跳转对根本不存在在输出结果到 MySQL 之前要把通道里的页面枚举列出来和跳转统计结果做一次外连接不存在的跳转对补 0否则报表系统读到的数据会出现大量缺行。3.3 JDBCHelper 批量写 MySQL 的几个隐藏坑JDBCHelper是整个平台的数据出口session 统计结果、转化率结果、热门商品结果都要通过它写进 MySQL。直接调用Statement.executeBatch()看起来没什么问题但实际跑批时掉过不少坑。第一个坑是批量插入条数过多导致内存溢出一般每批控制在 500 条左右用addBatch()累加到阈值再executeBatch()然后clearBatch()。第二个坑是 MySQL 连接长时间不释放会触发 wait_timeout连接池里的连接被服务端断开程序报Communications link failure需要定期用SELECT 1做心跳校验。第三个坑是事务控制JDBCHelper.executeBatch()之外还要包一层事务要么全部写入成功要么全部回滚避免报表数据读到一半的结果。public int[] insertBatch(String sql, ListObject[] paramsList) { int[] result null; Connection conn getConnection(); try { conn.setAutoCommit(false); PreparedStatement pstmt conn.prepareStatement(sql); for (Object[] params : paramsList) { for (int i 0; i params.length; i) { pstmt.setObject(i 1, params[i]); } pstmt.addBatch(); } result pstmt.executeBatch(); conn.commit(); pstmt.clearBatch(); } catch (SQLException e) { try { conn.rollback(); } catch (SQLException ex) { log.error(回滚失败, ex); } log.error(批量插入失败, e); } finally { closeConnection(conn); } return result; }批量操作的价值不只是减少网络往返更重要的是让写入过程整体可控。如果一条一条插几万条结果会拖死 MySQL 的线程池分批之后配合事务即使某批失败也能定位到具体范围。代码里把autoCommit关闭再手动提交是为了保证一批数据要么全部落库要么全部回滚不会出现报表里某次统计数据残了一半的情况。连接复用方面getConnection()内部建议使用 Druid 或 HikariCP 连接池而不是每次DriverManager.getConnection()后者在高并发写库时会频繁创建连接拖垮整个任务。4. 热门商品离线统计与广告流量实时统计的实现4.1 CategorySortKey 二次排序的设计思路热门商品离线统计的输入是用户点击、下单、支付三类行为数据输出是每个品类按点击次数优先、下单次数次之、支付次数最后排序的结果。这个排序规则涉及多个字段的比较直接使用 Java 默认排序做不到所以要自定义CategorySortKey实现Comparable接口。排序 Key 的字段顺序决定了比较的优先级实现里先比 clickCount相等再比 orderCount最后比 payCount。public class CategorySortKey implements ComparableCategorySortKey { private long clickCount; private long orderCount; private long payCount; Override public int compareTo(CategorySortKey o) { if (this.clickCount ! o.clickCount) { return (int) (this.clickCount - o.clickCount); } else if (this.orderCount ! o.orderCount) { return (int) (this.orderCount - o.orderCount); } else { return (int) (this.payCount - o.payCount); } } Override public boolean equals(Object obj) { if (!(obj instanceof CategorySortKey)) { return false; } CategorySortKey other (CategorySortKey) obj; return this.clickCount other.clickCount this.orderCount other.orderCount this.payCount other.payCount; } Override public int hashCode() { return (int) (clickCount * 31 orderCount * 17 payCount * 7); } }实现compareTo之后还要重写equals和hashCode原因在于sortByKey内部会触发数据混洗和聚合操作如果对象比较逻辑不一致排序结果会随机出错。排序的方向也有讲究Comparable返回正数表示当前对象排在后面如果想让点击量大的品类排前面需要返回相反数。实践里通常把返回值取反来得到降序而不是在 SQL 里依赖ORDER BY DESC因为 RDD 层面的排序更可控可以直接作用于已经聚合好的品类维度数据。热门商品的统计口径还要统一点击、下单、支付三张行为表的数据量差异很大通常在 join 之前先把三者的聚合结果单独算出来再通过品类 ID 关联避免大表 join 小表时产生大量 shuffle 数据。4.2 广告点击实时统计的窗口计算与状态维护广告流量实时统计是整套平台里唯一跑在 Spark Streaming 上的模块。数据源是模拟的广告点击流以 Kafka 为消息队列时Producer 端发送的每条消息包含时间戳、广告 ID、用户 ID 和点击位置。Streaming 端的数据处理链路是从 Kafka 拉取 DStream先用map解析出广告 ID然后用reduceByKeyAndWindow计算最近 60 秒内的点击量窗口滑动间隔设为 10 秒这样每 10 秒刷新一次最近一分钟的统计报表。JavaPairDStreamString, Long adClickDStream kafkaDStream .mapToPair(line - { String[] fields line._2().split(,); // 字段顺序: adId,userId,timestamp,position return new Tuple2(fields[0], 1L); }); JavaPairDStreamString, Long windowedClickCount adClickDStream .reduceByKeyAndWindow( (Long v1, Long v2) - v1 v2, (Long v1, Long v2) - v1 - v2, Durations.seconds(60), Durations.seconds(10) );reduceByKeyAndWindow的第二个函数是逆函数用于移除滑出窗口的数据这是它比reduceByKey之后再截取窗口高效的原因。使用逆函数时内存里只需要维护当前窗口的累计值每次滑动只需对新进入的数据做加和、对离开的数据做减法不需要重新计算整个窗口。但要注意逆函数的计算逻辑必须和加函数严格对称如果加函数里有非交换操作窗口结果会漂移。广告点击的实时统计还需要处理用户黑名单过滤判断逻辑通常在进入窗口计算之前完成从 Redis 或 MySQL 读取黑名单集合在 DStream 的filter里过滤掉黑名单用户避免恶意点击流量污染统计结果。4.3 实时任务的 checkpoint 与 Spark on YARN 提交Spark Streaming 任务的可靠性依赖 checkpoint它会保存 RDD 的血缘关系和 offset 信息。工程里把 checkpoint 目录配到 HDFS 上这样任务重启之后可以从最近一次保存的状态恢复而不是从头消费 Kafka 里的所有数据。checkpoint 的时间间隔一般取 batchDuration 的 5 到 10 倍例如 batchDuration 为 10 秒checkpoint 间隔设为 60 秒太频繁会引入大量 HDFS 写操作太稀疏会让恢复时的数据丢失窗口变大。实时任务和离线任务的提交方式也不同。离线任务用spark-submit一次性提交跑完就退出实时任务提交到 YARN 上之后需要常驻退出策略要配置为--supervise。使用spark on yarn提交时提交端只是把 jar 和依赖上传到集群实际执行在 NodeManager 的 container 里如果本地环境变量和集群不一致会在启动阶段报ClassNotFoundException。所以提交脚本里要把--jars指向 mysql-connector 和 kafka-clients 的依赖尽量用--deploy-mode cluster而不是 client 模式避免 driver 跑在提交机上、网络波动导致任务异常退出。实时任务上线前还要预估数据峰值把spark.streaming.kafka.maxRatePerPartition配好否则数据洪峰进来直接把内存打爆。5. 数据倾斜定位与两阶段聚合调优数据倾斜是这套平台跑批任务时最容易遇到、也最隐蔽的问题。场景非常典型某个热门品类在点击行为表里占了 60% 的数据量按品类聚合时所有数据都挤到同一个 task 上其他 task 十几秒跑完这一个 task 要跑十几分钟甚至 OOM。定位倾斜位置先到 Spark UI 上看 Stage 的耗时柱状图如果某一个 task 的 Shuffle Read 数据量明显大于其余 task就是数据倾斜。另外要看日志里有没有java.lang.OutOfMemoryError倾斜往往会导致单个 task 的堆内存被撑爆。两阶段聚合是解决按键聚合倾斜的首选方案。思路是先给每个 key 加一个随机前缀把原本一个 key 的数据打散到多个 task 上做局部聚合然后去掉前缀再做一次全局聚合。这样可以避免单个 task 处理过多数据代价是多了一次 shuffle。代码如下所示。JavaPairRDDString, Long fixedRDD skewedRDD // 第一阶段给 key 加随机前缀 .mapToPair(tuple - { String categoryId tuple._1(); Random random new Random(); int prefix random.nextInt(10); return new Tuple2(prefix _ categoryId, tuple._2()); }) .reduceByKey((v1, v2) - v1 v2) // 第二阶段去掉前缀还原真实 key .mapToPair(tuple - { String originalKey tuple._1().substring(tuple._1().indexOf(_) 1); return new Tuple2(originalKey, tuple._2()); }) .reduceByKey((v1, v2) - v1 v2);随机前缀的范围不宜太大10 到 50 之间比较合适。范围太小打散效果不明显范围太大会增加第一阶段的 shuffle 数据量和 reduceByKey 的调度开销。对于数据量在千万级以下的任务前缀范围取 10 就够了。两阶段聚合只适用于reduceByKey或groupByKey这类按键聚合场景。如果是 join 导致的数据倾斜比如大表关联小表可以走广播变量方案把小表用broadcast()发到每个 executor避免 shuffle。下面给出调优参数对照表覆盖常见瓶颈。问题场景调优参数建议值说明Shuffle 输出文件过多spark.sql.shuffle.partitions8 至 64根据数据量调整避免分区数过多导致调度开销Executor 单任务 OOMspark.executor.memory4G 至 8G配合spark.memory.fraction0.6使用小表 join 大表数据倾斜spark.sql.autoBroadcastJoinThreshold默认 10MB可调 100MB超过阈值再触发 shuffle join实时任务背压spark.streaming.backpressure.enabledtrue动态控制 Kafka 消费速率防止峰值打爆内存频繁 GC 导致 task 变慢spark.executor.extraJavaOptions-XX:UseG1GC大数据量任务建议 G1 替代默认回收器参数调完不要只看任务是否成功要看 Spark UI 里每个 Stage 的耗时分布是不是趋于均匀。调优验证的标准是最慢 task 的耗时不超过平均耗时的两倍Shuffle Read 的最大值和平均值接近GC 耗时不占 task 总耗时的 20% 以上。这套平台里如果 session 聚合阶段遇到倾斜同样可以用随机前缀方案处理而且模拟数据本身就是倾斜分布的直接拿这个案例做实验就能看到两阶段聚合前后的 task 耗时差异比在真实生产环境里调试安全得多。本文还有配套的精品资源点击获取
简介面向大数据开发与Spark学习者的电商用户行为分析平台项目提供完整源码与模拟数据集。项目覆盖用户session分析、页面单跳转化率统计、热门商品离线统计、广告流量实时统计四大业务模块涉及Spark Core、Spark SQL、Spark Streaming三套技术框架并融入数据倾斜处理、线上故障排查、性能调优等进阶经验。压缩包共236个文件以40个Java源码、184个XML配置为主另含6张PNG示意图整体大小仅1.23MB目录结构清晰便于对照学习。整套资源从需求分析、方案设计到编码测试与调优均有体现模拟数据可直接运行验证已有387人学习下载适合具备Spark基础、希望提升大数据项目实战能力的读者。1. 用户行为分析不是写几个 SQL 就能交付的接手这套电商用户行为分析平台之前我一度以为把访问日志导进 Hive写几条聚合 SQL 就能出数。真正跑起来才发现业务方要的不是一张表而是一连串连环问题今天新增了多少有效 sessionsession 时长的分布曲线长什么样从首页到商品详情页的转化掉了多少哪些品类值得上首页推荐广告点击流量在分钟级有没有异常波动。这些需求横跨离线批处理和实时流处理只靠 SQL 或者只靠一个计算框架都撑不住。这套基于 Spark 的电商用户行为分析大数据平台源码把四个业务模块串成了完整闭环用户 session 分析、页面单跳转化率统计、热门商品离线统计、广告流量实时统计。工程里既有 Spark Core 的 RDD 算子实战也有 Spark SQL 的结构化查询还有 Spark Streaming 的实时计算以及贯穿全程的数据倾斜与性能调优经验。适合有两到三个月 Spark 基础、想完整拆一个企业级项目的开发者也适合正在设计数据平台、需要参考模块划分和技术选型的一线工程师。读源码的价值不只是跑通流程而是看作者在需求分析、数据设计、编码实现、测试和调优这几个环节里到底做了哪些取舍。2. 平台分层与 Spark 三个子框架的边界2.1 MockData 打点逻辑与数据模型平台跑在模拟数据上所以先要读懂MockData.java里数据是怎么造出来的。模拟数据覆盖用户访问行为、用户基本信息和商品信息三类表其中用户访问行为表是核心字段包含 date、user_id、session_id、page_id、action_time、search_keyword、click_category_id、click_product_id、order_category_ids、order_product_ids、pay_category_ids、pay_product_ids。一个 session_id 代表用户一次连续的访问会话行为类型通过字段是否有值来区分比如同时有 click_category_id 和 click_product_id 说明是一次点击行为order_category_ids 非空说明产生了下单pay_category_ids 非空说明完成了支付。模拟数据生成要注意两个参数数据量和 session 的分布密度。一般做法是控制 user_id 在某个区间内随机比如 1 到 100 的用户但把 session_id 的基数放大到 10000 以上这样 groupBy session 时才会产生足够多的分组聚合统计才有意义。另一个关键是行为时间要递增不能出现同一个 session 里时间倒挂否则后续 session 时长和单跳转化率的计算结果都是错的。造数时倾向于把 70% 的行为集中在少数热门 session 上剩下 30% 均匀分布刻意制造一些倾斜等第四模块调优时就能直接看到效果。2.2 源码工程里的核心对象与职责项目工程结构不复杂核心类之间的协作关系却很清晰我在下表中整理了主要对象的职责边界。源码对象职责关键方法/属性MockData.java生成模拟访问行为、用户信息、商品信息数据generateUserVisitActionData()UserVisitAnalyze.java用户 session 分析主入口负责按时间区间聚合 session过滤有效 sessionmain()、sessionAggrStat()SessionAggrStat.java对 session 的访问时长、访问步长做分段统计calculate()CategorySortKey.java实现品类排序的二次排序 Key比较点击量、下单量、支付量compareTo()JDBCHelper.javaMySQL 写入工具封装批量插入、连接复用、事务控制insertBatch()DateUtils.java日期格式化、时间差计算用于 session 时长和单跳时间间隔getTimeDifference()SessionDetail.java保存 session 明细数据供聚合后的明细查询使用actionList模块之间通过 RDD 和 DataFrame 传递数据JDBCHelper是唯一的数据出口所有统计结果最终都落到 MySQL。CategorySortKey这个类是重点它不是普通排序而是让 Spark 在 reduceByKey 或 sortByKey 时按点击优先、其次下单、最后支付的顺序排这是电商热门商品统计里非常典型的排序需求。2.3 Spark Core、SQL 与 Streaming 的分工不是拍脑袋读这个项目最大的收获之一是搞清楚为什么同一套平台里要同时用 Spark Core、Spark SQL 和 Spark Streaming而不是图省事全用 RDD 一把梭。用户 session 分析这一步逻辑复杂需要先按 session_id 分组再对每组的访问时长和步长做条件过滤和分段这种逐条数据的分支判断用 RDD 算子写起来最直接。Spark Core 的 map、filter、reduceByKey 在处理这种非结构化、带大量业务分支的逻辑时灵活性和可控性比 DataFrame 高尤其是调试阶段可以随时打印中间结果。页面单跳转化率和热门商品统计的数据形态是规整的日志表过滤、关联、聚合的语义非常明确用 Spark SQL 最划算。同样一段取每个页面 PV 的逻辑DataFrame 的 groupBy 和 agg 写出来不到十行换成 RDD 要处理各种序列化和类型转换。这里要注意一个容易踩的坑Spark SQL 的spark.sql.shuffle.partitions默认值是 200如果数据量只有几百 MB200 个分区反而会让每个 task 处理的数据量过小shuffle 文件过多调度开销变大。小数据集跑批任务时我一般把它调成数据量 / 128MB的倍数比如 1GB 的数据量设置为 8 到 16 个分区。Spark Streaming 的任务是广告流量实时统计它和离线任务的边界在于时间和状态。离线任务处理的是今天之前的所有数据实时任务处理的是最近几秒进来的数据。如果广告点击流也要做历史累计则必须在 Streaming 里维护跨批次的状态这就要用到 updateStateByKey 或 mapWithState。三个框架在项目里不是堆砌而是根据数据的时效性和计算逻辑的复杂性分工数据越规整、时效性要求越高就越往 SQL 和 Streaming 靠逻辑分支越多、数据越杂越需要 Core 的算子能力兜底。3. Session 聚合与页面单跳转化率的实现细节3.1 Session 切分逻辑与步长时长分段统计用户 session 分析的起点是从原始访问行为数据中切分出一个个 session。切分规则通常是同一个 session_id 内的行为时间间隔不超过 30 分钟超过 30 分钟视为另一次会话。工程里的实现方式是先读取原始访问行为数据过滤出指定时间范围内的记录然后按 session_id 做 groupByKey再对每组内部的 action_time 排序。排序这一步容易被忽略因为 groupByKey 之后 RDD 的分组内部并不保证有序如果直接遍历计算步长会出现时序错乱。JavaPairRDDString, IterableUserVisitAction sessionActions actionRDD.groupByKey(); JavaPairRDDString, SessionAggrStat sessionStatRDD sessionActions .mapToPair(tuple - { String sessionId tuple._1(); IterableUserVisitAction actions tuple._2(); ListUserVisitAction actionList new ArrayList(); for (UserVisitAction action : actions) { actionList.add(action); } // 按访问时间排序保证步长和时长的计算顺序正确 actionList.sort(Comparator.comparing(UserVisitAction::getActionTime)); long visitLength 0L; long stepLength 0L; long startTime 0L; long endTime 0L; for (int i 0; i actionList.size(); i) { UserVisitAction action actionList.get(i); if (i 0) { startTime DateUtils.parseToLong(action.getActionTime()); } if (i actionList.size() - 1) { endTime DateUtils.parseToLong(action.getActionTime()); } stepLength; } visitLength (endTime - startTime) / 1000; return new Tuple2(sessionId, new SessionAggrStat(sessionId, visitLength, stepLength)); });这段代码完成的是 session 步长和时长的初步计算。先对每个 session 内的行为按时间排序然后遍历一次拿到第一条和最后一条行为的时间戳算出访问时长同时累加行为条数作为访问步长。这里有个关键细节访问时长是按秒算还是按毫秒算直接决定了后面分段统计的边界值。项目里统一按秒计算因为业务方看的是分钟和小时的分布秒级精度足够。拿到每个 session 的时长和步长之后SessionAggrStat负责做分段统计。分段的方式常见的是区间映射访问时长分成 0 到 3 秒、4 到 10 秒、11 到 30 秒、31 到 60 秒、1 到 3 分钟、3 分钟以上这几档访问步长分成 1 到 3 步、4 到 6 步、7 到 9 步、10 到 30 步、30 步以上。实现时避免用 if-else 堆一串边界判断更优雅的方案是先把边界值放进一个数组遍历数组用二分查找定位区间下标再用一个累加数组计数。3.2 页面单跳转化率的计算链路页面单跳转化率是电商分析里最容易被业务方追问的指标之一。它衡量的是用户从页面 A 跳转到页面 B 的概率比如从首页跳到搜索结果页的转化率是多少从搜索结果页跳到商品详情页的转化率是多少。计算链路分两步第一步统计每个页面被访问的总次数PV第二步统计从页面 A 跳转到页面 B 的次数。-- 第一步统计每个页面的 PV SELECT page_id, COUNT(*) AS pv FROM user_visit_action WHERE date 2024-11-20 GROUP BY page_id;第二步的跳转次数统计不能直接用 SQL 的 GROUP BY 完成因为需要识别相邻的访问行为。工程里的实现方式是先按 session_id 分组在每个 session 内部把行为按时间排序然后生成相邻页面对。这里用到了 RDD 的mapPartitions在分区内处理排序后的列表遍历时每遇到相邻两次访问行为就输出一个以页面A_页面B为 key 的计数。JavaPairRDDString, Long pageJumpPairRDD sessionActions.flatMapToPair( (FlatMapFunctionTuple2String, IterableUserVisitAction, Tuple2String, Long) tuple - { ListUserVisitAction actions new ArrayList(); tuple._2().forEach(actions::add); actions.sort(Comparator.comparing(UserVisitAction::getActionTime)); ListTuple2String, Long list new ArrayList(); for (int i 0; i actions.size() - 1; i) { String fromPage String.valueOf(actions.get(i).getPageId()); String toPage String.valueOf(actions.get(i 1).getPageId()); list.add(new Tuple2(fromPage _ toPage, 1L)); } return list.iterator(); });生成跳转对之后用reduceByKey累加每个页面A_页面B的总次数。转化率的计算不能只看跳转次数必须拿跳转次数除以起始页面的总 PV这样才能消除页面本身流量大小的影响。还有一个工程上的迭代点页面集合是固定的很多跳转对根本不存在在输出结果到 MySQL 之前要把通道里的页面枚举列出来和跳转统计结果做一次外连接不存在的跳转对补 0否则报表系统读到的数据会出现大量缺行。3.3 JDBCHelper 批量写 MySQL 的几个隐藏坑JDBCHelper是整个平台的数据出口session 统计结果、转化率结果、热门商品结果都要通过它写进 MySQL。直接调用Statement.executeBatch()看起来没什么问题但实际跑批时掉过不少坑。第一个坑是批量插入条数过多导致内存溢出一般每批控制在 500 条左右用addBatch()累加到阈值再executeBatch()然后clearBatch()。第二个坑是 MySQL 连接长时间不释放会触发 wait_timeout连接池里的连接被服务端断开程序报Communications link failure需要定期用SELECT 1做心跳校验。第三个坑是事务控制JDBCHelper.executeBatch()之外还要包一层事务要么全部写入成功要么全部回滚避免报表数据读到一半的结果。public int[] insertBatch(String sql, ListObject[] paramsList) { int[] result null; Connection conn getConnection(); try { conn.setAutoCommit(false); PreparedStatement pstmt conn.prepareStatement(sql); for (Object[] params : paramsList) { for (int i 0; i params.length; i) { pstmt.setObject(i 1, params[i]); } pstmt.addBatch(); } result pstmt.executeBatch(); conn.commit(); pstmt.clearBatch(); } catch (SQLException e) { try { conn.rollback(); } catch (SQLException ex) { log.error(回滚失败, ex); } log.error(批量插入失败, e); } finally { closeConnection(conn); } return result; }批量操作的价值不只是减少网络往返更重要的是让写入过程整体可控。如果一条一条插几万条结果会拖死 MySQL 的线程池分批之后配合事务即使某批失败也能定位到具体范围。代码里把autoCommit关闭再手动提交是为了保证一批数据要么全部落库要么全部回滚不会出现报表里某次统计数据残了一半的情况。连接复用方面getConnection()内部建议使用 Druid 或 HikariCP 连接池而不是每次DriverManager.getConnection()后者在高并发写库时会频繁创建连接拖垮整个任务。4. 热门商品离线统计与广告流量实时统计的实现4.1 CategorySortKey 二次排序的设计思路热门商品离线统计的输入是用户点击、下单、支付三类行为数据输出是每个品类按点击次数优先、下单次数次之、支付次数最后排序的结果。这个排序规则涉及多个字段的比较直接使用 Java 默认排序做不到所以要自定义CategorySortKey实现Comparable接口。排序 Key 的字段顺序决定了比较的优先级实现里先比 clickCount相等再比 orderCount最后比 payCount。public class CategorySortKey implements ComparableCategorySortKey { private long clickCount; private long orderCount; private long payCount; Override public int compareTo(CategorySortKey o) { if (this.clickCount ! o.clickCount) { return (int) (this.clickCount - o.clickCount); } else if (this.orderCount ! o.orderCount) { return (int) (this.orderCount - o.orderCount); } else { return (int) (this.payCount - o.payCount); } } Override public boolean equals(Object obj) { if (!(obj instanceof CategorySortKey)) { return false; } CategorySortKey other (CategorySortKey) obj; return this.clickCount other.clickCount this.orderCount other.orderCount this.payCount other.payCount; } Override public int hashCode() { return (int) (clickCount * 31 orderCount * 17 payCount * 7); } }实现compareTo之后还要重写equals和hashCode原因在于sortByKey内部会触发数据混洗和聚合操作如果对象比较逻辑不一致排序结果会随机出错。排序的方向也有讲究Comparable返回正数表示当前对象排在后面如果想让点击量大的品类排前面需要返回相反数。实践里通常把返回值取反来得到降序而不是在 SQL 里依赖ORDER BY DESC因为 RDD 层面的排序更可控可以直接作用于已经聚合好的品类维度数据。热门商品的统计口径还要统一点击、下单、支付三张行为表的数据量差异很大通常在 join 之前先把三者的聚合结果单独算出来再通过品类 ID 关联避免大表 join 小表时产生大量 shuffle 数据。4.2 广告点击实时统计的窗口计算与状态维护广告流量实时统计是整套平台里唯一跑在 Spark Streaming 上的模块。数据源是模拟的广告点击流以 Kafka 为消息队列时Producer 端发送的每条消息包含时间戳、广告 ID、用户 ID 和点击位置。Streaming 端的数据处理链路是从 Kafka 拉取 DStream先用map解析出广告 ID然后用reduceByKeyAndWindow计算最近 60 秒内的点击量窗口滑动间隔设为 10 秒这样每 10 秒刷新一次最近一分钟的统计报表。JavaPairDStreamString, Long adClickDStream kafkaDStream .mapToPair(line - { String[] fields line._2().split(,); // 字段顺序: adId,userId,timestamp,position return new Tuple2(fields[0], 1L); }); JavaPairDStreamString, Long windowedClickCount adClickDStream .reduceByKeyAndWindow( (Long v1, Long v2) - v1 v2, (Long v1, Long v2) - v1 - v2, Durations.seconds(60), Durations.seconds(10) );reduceByKeyAndWindow的第二个函数是逆函数用于移除滑出窗口的数据这是它比reduceByKey之后再截取窗口高效的原因。使用逆函数时内存里只需要维护当前窗口的累计值每次滑动只需对新进入的数据做加和、对离开的数据做减法不需要重新计算整个窗口。但要注意逆函数的计算逻辑必须和加函数严格对称如果加函数里有非交换操作窗口结果会漂移。广告点击的实时统计还需要处理用户黑名单过滤判断逻辑通常在进入窗口计算之前完成从 Redis 或 MySQL 读取黑名单集合在 DStream 的filter里过滤掉黑名单用户避免恶意点击流量污染统计结果。4.3 实时任务的 checkpoint 与 Spark on YARN 提交Spark Streaming 任务的可靠性依赖 checkpoint它会保存 RDD 的血缘关系和 offset 信息。工程里把 checkpoint 目录配到 HDFS 上这样任务重启之后可以从最近一次保存的状态恢复而不是从头消费 Kafka 里的所有数据。checkpoint 的时间间隔一般取 batchDuration 的 5 到 10 倍例如 batchDuration 为 10 秒checkpoint 间隔设为 60 秒太频繁会引入大量 HDFS 写操作太稀疏会让恢复时的数据丢失窗口变大。实时任务和离线任务的提交方式也不同。离线任务用spark-submit一次性提交跑完就退出实时任务提交到 YARN 上之后需要常驻退出策略要配置为--supervise。使用spark on yarn提交时提交端只是把 jar 和依赖上传到集群实际执行在 NodeManager 的 container 里如果本地环境变量和集群不一致会在启动阶段报ClassNotFoundException。所以提交脚本里要把--jars指向 mysql-connector 和 kafka-clients 的依赖尽量用--deploy-mode cluster而不是 client 模式避免 driver 跑在提交机上、网络波动导致任务异常退出。实时任务上线前还要预估数据峰值把spark.streaming.kafka.maxRatePerPartition配好否则数据洪峰进来直接把内存打爆。5. 数据倾斜定位与两阶段聚合调优数据倾斜是这套平台跑批任务时最容易遇到、也最隐蔽的问题。场景非常典型某个热门品类在点击行为表里占了 60% 的数据量按品类聚合时所有数据都挤到同一个 task 上其他 task 十几秒跑完这一个 task 要跑十几分钟甚至 OOM。定位倾斜位置先到 Spark UI 上看 Stage 的耗时柱状图如果某一个 task 的 Shuffle Read 数据量明显大于其余 task就是数据倾斜。另外要看日志里有没有java.lang.OutOfMemoryError倾斜往往会导致单个 task 的堆内存被撑爆。两阶段聚合是解决按键聚合倾斜的首选方案。思路是先给每个 key 加一个随机前缀把原本一个 key 的数据打散到多个 task 上做局部聚合然后去掉前缀再做一次全局聚合。这样可以避免单个 task 处理过多数据代价是多了一次 shuffle。代码如下所示。JavaPairRDDString, Long fixedRDD skewedRDD // 第一阶段给 key 加随机前缀 .mapToPair(tuple - { String categoryId tuple._1(); Random random new Random(); int prefix random.nextInt(10); return new Tuple2(prefix _ categoryId, tuple._2()); }) .reduceByKey((v1, v2) - v1 v2) // 第二阶段去掉前缀还原真实 key .mapToPair(tuple - { String originalKey tuple._1().substring(tuple._1().indexOf(_) 1); return new Tuple2(originalKey, tuple._2()); }) .reduceByKey((v1, v2) - v1 v2);随机前缀的范围不宜太大10 到 50 之间比较合适。范围太小打散效果不明显范围太大会增加第一阶段的 shuffle 数据量和 reduceByKey 的调度开销。对于数据量在千万级以下的任务前缀范围取 10 就够了。两阶段聚合只适用于reduceByKey或groupByKey这类按键聚合场景。如果是 join 导致的数据倾斜比如大表关联小表可以走广播变量方案把小表用broadcast()发到每个 executor避免 shuffle。下面给出调优参数对照表覆盖常见瓶颈。问题场景调优参数建议值说明Shuffle 输出文件过多spark.sql.shuffle.partitions8 至 64根据数据量调整避免分区数过多导致调度开销Executor 单任务 OOMspark.executor.memory4G 至 8G配合spark.memory.fraction0.6使用小表 join 大表数据倾斜spark.sql.autoBroadcastJoinThreshold默认 10MB可调 100MB超过阈值再触发 shuffle join实时任务背压spark.streaming.backpressure.enabledtrue动态控制 Kafka 消费速率防止峰值打爆内存频繁 GC 导致 task 变慢spark.executor.extraJavaOptions-XX:UseG1GC大数据量任务建议 G1 替代默认回收器参数调完不要只看任务是否成功要看 Spark UI 里每个 Stage 的耗时分布是不是趋于均匀。调优验证的标准是最慢 task 的耗时不超过平均耗时的两倍Shuffle Read 的最大值和平均值接近GC 耗时不占 task 总耗时的 20% 以上。这套平台里如果 session 聚合阶段遇到倾斜同样可以用随机前缀方案处理而且模拟数据本身就是倾斜分布的直接拿这个案例做实验就能看到两阶段聚合前后的 task 耗时差异比在真实生产环境里调试安全得多。本文还有配套的精品资源点击获取
