★ FEATURED ARTICLE 简介面向大数据初学者的Spark电子书平台数据分析完整工程围绕下载量统计、评分计算、阅读时段分析三类典型实时业务场景展开。项目以Scala语言实现覆盖各标签总点击下载量实时统计、标签内下载Top10榜单、每本电子书评分动态更新以及0:00-6:00、6:00-12:00、12:00-18:00、18:00-24:00四个时段在线人数监控等核心模块可直接作为课程设计、毕业设计或Spark入门实战参考。压缩包共51个文件主要包括36个scala源码、9个xml工程配置、2个docx项目文档、data样例数据与json结果文件包体仅1.67MB便于快速下载与本地运行。当前已有117人学习下载。附带u.data等测试数据及click_data.json输出部署后即可观察各模块统计效果项目文档还能帮助理解Spark实时分析的设计思路与实现要点。1. 基于Spark的电子书平台数据分析设计与实现为什么说这是最有性价比的大数据练手项目做电子书平台的都知道每天后台攒下的不只是订单还有一长串用户行为谁在搜书、谁点了收藏、谁把某本书读到第三章又退出。这些数据单个看没什么堆到千万级之后Excel 会崩MySQL 单表聚合也开始变慢。基于 Spark 的电子书平台数据分析设计与实现就是拿 Spark 把这种体量的日志和数据跑成运营能直接用的日报、画像和榜单。整套工程拆开看是三件事设计数据模型、写 Spark 分析任务、把结果落到报表和可视化里。“源码项目文档”的交付形式决定了它不是讲概念的 PPT而是一套能照着搭、能跑通、能讲清每个参数为什么这么配的完整样板。无论你是在为大数据岗位面试攒项目还是想在离线数据分析方向沉淀一套可复用的技术模板这个题目都值得投入。2. 数据设计先行Spark读取JSON的方式和电子书平台的表结构数据分析项目最先翻车的往往不是代码而是数据源没设计好。先花一整章把数据层讲透后面所有分析才能站得住。这里按“日志怎么进”、“表怎么建”、“代码怎么摆”的顺序展开。2.1 Spark中读取JSON埋点日志的字段设计与read.json用法电子书平台的前端埋点一般直接产生 JSON 日志后端也可能把访问日志落成 JSON 文件。JSON 的好处是方便加字段不需要提前建表坏处是垃圾字段多、schema 不稳定。Spark 读取 JSON 没有黑科技一行spark.read.json(path)就能读出一个 DataFrame但生产环境不能这么裸奔。来看一个完整的最小代码from pyspark.sql import SparkSession from pyspark.sql.types import ( StructType, StructField, StringType, IntegerType, TimestampType ) spark SparkSession.builder \ .appName(ebook_log_read) \ .config(spark.sql.adaptive.enabled, true) \ .getOrCreate() schema StructType([ StructField(user_id, StringType(), True), StructField(book_id, StringType(), True), StructField(chapter_id, IntegerType(), True), StructField(event_time, TimestampType(), True), StructField(action, StringType(), True), StructField(read_duration, IntegerType(), True), StructField(device_type, StringType(), True), StructField(channel, StringType(), True) ]) logs_df spark.read \ .option(ignoreCorruptFiles, true) \ .schema(schema) \ .json(hdfs:///data/ebook/logs/dt2025-03-15/*.json) logs_df.printSchema() logs_df.show(5, truncateFalse)这里两个关键点手动给 schema以及把spark.sql.adaptive.enabled打开。手动指定 schema 是为了防 Spark 自动推断时把某个文件里的read_duration推断成字符串导致后面的 sum 直接变成 null。ignoreCorruptFiles的作用是跳过损坏文件而不是让整个任务红掉这个参数在老集群上务必显式写。参数方面如果拿到的 JSON 每行是一个对象用默认的multiLinefalse就行如果整个文件是一个大的 JSON 数组需要加.option(multiLine, true)。还有一个容易忽略的点前端埋点常把页码、章节信息放在嵌套对象里分析时用col(page_info.chapter_id)取子字段或者用explode展开数组。字段名里可以有下划线但不要用点号做字段分隔符否则每次引用都要反引号血泪经验。2.2 表模型设计用户维表、书籍维表、阅读行为事实表数据分析的第一步是把业务关系摆正。电子书平台常见的建模方式是“两张维表 一张事实表”也就是星型模型。维表变化慢事实表每天会新增几百万行。表表名关键字段更新频率用户维表dim_useruser_id, register_time, gender, age_range, membership_level每日全量书籍维表dim_bookbook_id, book_name, author, category, word_count, price, shelf_status每日全量阅读行为事实表dwd_ebook_read_loguser_id, book_id, event_time, action, read_duration, device_type, channel, dt每日增量落地时我一般用 Spark SQL 把表结构和字段约束写清楚让任务每天用INSERT OVERWRITE刷新当日分区。CREATE TABLE IF NOT EXISTS dwd_ebook_read_log ( user_id STRING COMMENT 用户ID, book_id STRING COMMENT 书籍ID, event_time TIMESTAMP COMMENT 事件时间, action STRING COMMENT read/collect/search/config, read_duration INT COMMENT 阅读时长(秒), device_type STRING COMMENT pc/app/h5, channel STRING COMMENT 渠道 ) USING parquet PARTITIONED BY (dt STRING COMMENT 分区日期如2025-03-15);这张表后续所有分析都从它出数。事实表按天做分区是因为电子书平台的分析基本是日报一天一个分区既好管理也方便重跑某一天。USING parquet是重点Spark 读取 Parquet 比 JSON 快得多Parquet 自带列裁剪分析只需要user_id、book_id、read_duration三列时不会把整个文件全部搬进内存。2.3 最小项目目录结构源码和项目文档该怎么摆既然标题是“源码项目文档”代码组织就不能是单个 py 文件。见过太多项目把 ETL 和分析全写在一个 notebook 里跑通一次容易想重跑某一天时就麻了。我会按下面这个结构组织边界清楚。ebook-spark-analysis/ ├── src/ │ ├── main/python/ │ │ ├── parsing/ # 原始日志读取与schema定义 │ │ ├── etl/ # 清洗、去重、维度关联 │ │ ├── analysis/ # 日活、热榜、画像相关计算 │ │ └── export/ # 导出MySQL │ ├── main/resources/ │ │ └── application.conf # 数据库连接、路径、时间参数 ├── scripts/ │ └── submit.sh # spark-submit 跑批脚本 ├── docs/ │ ├── 01_需求文档.md # 指标口径、数据流图 │ └── 02_设计文档.md # 表模型、调度说明、参数说明 └── README.mdparsing 只负责“读出 DataFrame”etl 只负责“清洗产出 dwd 表”analysis 只负责“从 dwd 算指标”export 只负责“写出去”。这样改清洗逻辑不用动分析代码改输出方式也不用重跑整个链路。docs 里最该写清楚三件事拿什么跑spark-submit 参数、跑哪一天的数据时间参数、结果在哪看MySQL 表名和口径。项目文档写得越像操作手册这个项目的复现成本越低面试讲的时候也越有底气。3. Spark分析主流程从数据清洗到核心指标计算数据模型建好之后进入 Spark 分析核心环节。这一章的代码可以直接抄进你的项目里但一定要看我标出来的参数说明因为同一个写法在不同集群上表现可能差很远。3.1 数据清洗过滤脏数据、去重、统一时间口径电子书平台日志最常见的脏数据有三种user_id 为空、book_id 不存在、event_time 不在当天分区里。清洗规则不会太复杂但要把所有条件落在一个函数里避免散落各处。from pyspark.sql import functions as F def transform_read_log(raw_df): return (raw_df .filter(F.col(user_id).isNotNull() F.col(book_id).isNotNull() F.col(event_time).isNotNull()) .filter(F.col(event_time).between( 2025-03-15 00:00:00, 2025-03-15 23:59:59)) .dropDuplicates([user_id, book_id, chapter_id, event_time]) .withColumn(dt, F.to_date(F.col(event_time))))逻辑说明第一个 filter 去掉空主键第二个 filter 把时间窗固定到分析当天。dropDuplicates 按“同一用户对同一本书的同一章节在同一个时间点”去重。注意这里带上了 chapter_id目的是保留用户在同一个章节的多次连续阅读记录而不是把整本书的阅读记录合并成一条。withColumn生成 dt 字段后续所有分区裁剪都用它。参数说明event_time.between对 timestamp 类型是闭区间所以结束时间要写当天的 23:59:59不要写第二天 00:00:00否则会包进第二天零点整的数据。这个边界我至少踩过两次每次都是日报比业务方手算多几条记录最后定位到这里。3.2 核心指标UV、PV、阅读时长聚合覆盖大部分需求的核心指标是日活不重复用户数UV、页面浏览数PV和总阅读时长。Spark 里countDistinct是精确去重在数据量大时很贵如果是看板类允许偏差的场景用approx_count_distinct可以省不少 shuffle。daily_stats (clean_df .groupBy(dt) .agg( F.countDistinct(user_id).alias(uv), F.count(*).alias(pv), F.sum(read_duration).alias(total_read_seconds), F.avg(read_duration).alias(avg_read_seconds) ) .orderBy(dt)) daily_stats.show()这里要关注的是 Spark 在计算countDistinct时的内存行为。数据量到几亿行distinct 会触发大 shuffle如果遇到某本爆款书占了 80% 日志一个 executor 的堆内存很容易被打满。这时候先考虑approx_count_distinct或者把倾斜键单独拎出来算完再合并。另外sum(read_duration)之前要确认字段是整数类型否则结果全是 null这种问题最常出现在 JSON 推断出字符串列之后。书籍热度的计算会增加一个维度按书聚合book_rank (clean_df .groupBy(book_id, dt) .agg( F.countDistinct(user_id).alias(readers), F.sum(read_duration).alias(total_read_seconds) ) .filter(F.col(readers) 10) .orderBy(F.col(total_read_seconds).desc()) .limit(100)) book_rank.write.parquet(hdfs:///result/ebook/book_rank/dt2025-03-15)filter(readers 10)是常见套路过滤掉只有一两个人读的长尾书让榜单聚焦在真实内容上。limit(100) 只在最终结果取 100 本不会影响前面的聚合。写到 Parquet 按日期分区后续可视化直接读目录不用再算一遍。3.3 用户画像标签行为权重打分用户画像听起来玄学落到代码里其实就是给行为打权重再聚合。电子书平台的标签不需要实时推荐那样的复杂度离线打分足够支撑运营。规则是阅读行为按分钟数计分收藏一次计 2 分搜索一次计 1 分按用户汇总出对每个书籍类别的兴趣分。category_score (clean_df .join(F.broadcast(dim_book), book_id, left) .groupBy(user_id, category) .agg( F.sum( F.when(F.col(action) read, F.col(read_duration) / 60) .when(F.col(action) collect, 2.0) .when(F.col(action) search, 1.0) .otherwise(0.0) ).alias(category_score) ) .filter(F.col(category).isNotNull())) category_score.show(20)关键在F.when / .otherwise链式写法把一个阅读事件按动作类型换算成可累加的分数。read_duration以秒为单位除以 60 变分钟解决“读了一篇长文”和“点开 3 秒就退出”的行为分值差异。用F.broadcast(dim_book)是因为书籍维表通常只有几十万行广播到每个 executor 后 join 不会产生 shuffle这是常用的性能优化。3.4 一个Spark数据分析案例的扩展用ALS做简单的书单推荐如果想让项目文档里多一个“个性化推荐”模块最务实的是用 Spark MLlib 的 ALS 做协同过滤而不是自己写余弦相似度。ALS 对“用户—书籍—交互分数”矩阵天然友好封装也完整。from pyspark.ml import Pipeline from pyspark.ml.feature import StringIndexer from pyspark.ml.recommendation import ALS user_idx StringIndexer(inputColuser_id, outputColuser_idx) book_idx StringIndexer(inputColbook_id, outputColbook_idx) pipeline Pipeline(stages[user_idx, book_idx]) interaction_df (clean_df .groupBy(user_id, book_id) .agg(F.sum(F.col(read_duration) / 60).alias(interact_score)) .filter(F.col(interact_score) 1)) data pipeline.fit(interaction_df).transform(interaction_df) als ALS( maxIter10, regParam0.05, userColuser_idx, itemColbook_idx, ratingColinteract_score, coldStartStrategydrop ) model als.fit(data) recs model.recommendForAllUsers(5)interact_score过滤掉阅读不足 1 分钟的弱交互减少噪音。coldStartStrategydrop必须设置否则对新用户或新书会输出 null写表时直接报错。ALS 训练结束后recommendForAllUsers(5)返回每个用户 Top5但开销不小实际跑批通常只对近 7 天活跃用户做推荐而不是全量用户。这个模块放进 spark数据分析案例里是很自然的扩展不复杂但能把项目从“只会 groupBy”提升到“能跑机器学习”的层次。4. 结果落地MySQL导出、Python可视化和定时跑批分析结果不落库项目文档写得再漂亮也白搭。这一章处理“数据从哪里出去”的问题也是把 Spark 作业变成稳定服务的关键一步。4.1 用JDBC写回MySQL写入模式与批次参数把 Spark 聚合结果写回 MySQL最直接的是 DataFrame 的jdbc方法。下面是每日报表的落库代码import os daily_stats.write \ .mode(overwrite) \ .option(truncate, true) \ .jdbc(urljdbc:mysql://localhost:3306/ebook_analysis, tabledaily_report, properties{ user: root, password: os.environ.get(MYSQL_PASSWORD, ), driver: com.mysql.cj.jdbc.Driver, rewriteBatchedStatements: true, batchsize: 5000 })逻辑说明mode(overwrite)配合truncatetrue在 MySQL 端先清空目标表再写入适合每日全量结果表。这样重跑任务时 MySQL 不会因为残留旧数据产生重复记录。daily_stats只有几千行写库很快如果是要写几亿行明细不建议这个做法overwrite会很慢应该先落临时目录再 LOAD DATA。参数说明rewriteBatchedStatementstrue是 MySQL 批量写入的经典参数能把多个 insert 合并成一次网络往返。batchsize默认 1000调到 5000 通常更快但设过大会让 MySQL 端内存压力上升。密码不要写死在代码里从环境变量读最后由 spark-submit 注入。4.2 用Python做数据分析与可视化画趋势图Spark 适合算数不适合画图。最终呈现在看板上一般是把结果表拉回 Python用 pandas 加 matplotlib 处理。不要想着 Spark 自带图表也别在集群节点上跑带 GUI 的 Python那会把人折腾死。import os import pandas as pd import matplotlib.pyplot as plt from sqlalchemy import create_engine password os.environ.get(MYSQL_PASSWORD, ) engine create_engine(fmysqlpymysql://root:{password}localhost:3306/ebook_analysis) report pd.read_sql( SELECT dt, uv, pv, total_read_seconds FROM daily_report ORDER BY dt, engine) report[dt] pd.to_datetime(report[dt]) fig, ax plt.subplots(figsize(12, 5)) ax.plot(report[dt], report[uv], labelUV, markero) ax.plot(report[dt], report[pv], labelPV, markero) ax.set_title(电子书平台日活与浏览量趋势) ax.legend() plt.show()这个做法的前提是 daily_report 数据量不大一次性read_sql没问题。真正的生产系统里看板后端通常只查最近 30 天数据或者用接口返回给前端图表。本地做数据分析与可视化时这一套足够。补充一个容易踩的细节Spark 写进 MySQL 的dt字段是字符串类型Python 侧要主动转成 datetime否则 matplotlib 会把字符串当类别处理横轴顺序会乱。另外total_read_seconds单位是秒展示时最好在 Python 里除以 3600 变成小时不要直接在数据库里改字段语义。4.3 Spark搭建教程里必带的submit脚本与定时调度集群上的 Spark 作业不能像本地开发那样在 IDE 里点运行都要走spark-submit。我平时把跑批命令写成脚本放在scripts/submit.sh新人照着目录文档就能跑。#!/usr/bin/env bash set -euo pipefail DT${1:-$(date -d yesterday %F)} APP_JAR/opt/ebook-analysis/ebook-analysis.jar spark-submit \ --master yarn \ --deploy-mode cluster \ --name ebook_analysis_daily_${DT} \ --class com.example.EbookAnalysis \ --driver-memory 2g \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 4 \ --files /opt/ebook-analysis/conf/application.conf#application.conf \ ${APP_JAR} \ --dt ${DT}脚本里的每个参数都有含义--master yarn表示走 YARN 集群调度--deploy-mode cluster让 Driver 也在集群里运行本地客户端提交后即可退出--driver-memory要结合collect的量来设如果只在 Driver 上做了少量小结果展示2g 足够executor-memory4g、executor-cores2 是很多离线任务的起步配置。真正调优时去看 Spark UI 的 Executors 页如果 GC 时间占比高就加内存如果 CPU 利用率低就减少 core 数或增加并发。定时调度日常用 cron 就能满足30 2 * * * /opt/ebook-analysis/scripts/submit.sh $(date -d yesterday %F) /var/log/ebook-spark.log 21Spark 任务跑 30 分钟还是 3 个小时和资源队列、数据量强相关。cron 只是“触发”不能兜底失败重跑所以脚本里set -euo pipefail很有必要任何一步失败就非零退出日志才会有明确报错。更工程化一些可以接 DolphinScheduler 或 Airflow但小项目用 cron 加日志足够。5. 避坑清单基于Spark的电子书数据分析项目的5个高频问题下面这些坑都是我实际跑这类项目撞过的每一条按“现象、原因、解决”三个关键点写可以直接当排查手册用。5.1 现象读入大量JSON小文件时任务卡在FileScan阶段现象Spark 读取一天的 JSON 日志日志量只有 2GB但文件数有上万个任务在 FileScan 阶段就花了 40 分钟。原因每个小文件都会产生一个 partitionSpark 调度上万个 partition 的开销远大于数据处理本身。解决先跑一次小文件合并任务把 JSON 统一转成 Parquet 并按日期分区spark.read.schema(schema).json(hdfs:///data/ebook/logs/2025-03-15/*.json) \ .coalesce(50) \ .write.mode(overwrite) \ .partitionBy(dt) \ .parquet(hdfs:///data/ebook/dwd/read_log)这里coalesce(50)把结果控制在 50 个文件左右不是越大越好。partitionBy(dt)让后续查询可以自动做分区裁剪只读当天目录。如果原始文件已经是整天的目录优化后第二次跑同样指标通常快 5 倍以上。5.2 现象Executor lost / OOM容器被YARN杀掉现象作业在聚合阶段突然失败日志显示某个 executor 的 container killed或者java.lang.OutOfMemoryError。原因多数是数据倾斜加内存配置不对。比如某个热门小说占当天 80% 的阅读日志按 book_id groupBy 时处理那本书的任务全部落在同一个 executor 上堆被撑爆。解决先在 Spark UI 的 Stage 详情里找到 skew 的 task再把倾斜的 key 加盐salt F.col(book_id).cast(string) _ F.rand().cast(int) skewed_clean clean_df.withColumn(salt_book_id, salt) part_agg (skewed_clean .groupBy(salt_book_id) .agg(F.countDistinct(user_id).alias(readers), F.sum(read_duration).alias(total_read_seconds))) # 第二遍聚合去掉盐还原成真实book_id final (part_agg .withColumn(book_id, F.regexp_replace(salt_book_id, _\d$, )) .groupBy(book_id) .agg(F.sum(readers).alias(readers), F.sum(total_read_seconds).alias(total_read_seconds)))加盐后同一本书的数据会分散到 10 个 key 上每个 reducer 只处理 1/10 数据但还要再做一次去盐聚合把结果还原。更稳的做法是把spark.sql.shuffle.partitions调大并且给 executor 的spark.memory.overhead留足余量比如--conf spark.memory.overhead1g。Spark 内存配置有时候就是玄学看到 OOM 先加 overhead再调 shuffle 分区。5.3 现象本地模式跑通Spark集群搭建好后却跑不出结果现象代码在--master local[2]下跑得飞快提交到集群就报错错误千奇百怪比如找不到 MySQL 驱动、读不到 application.conf。原因本地模式自动把代码和配置放在同一个 JVM而 YARN 集群模式里代码在远端容器执行driver 和 executor 拿不到之前的 classpath 和文件。解决提交时显式带上资源文件spark-submit ... \ --files /opt/ebook-analysis/conf/application.conf#application.conf \ --conf spark.executor.extraJavaOptions-Dconfig.fileapplication.conf \ --jars /opt/ebook-analysis/mysql-connector-j-8.0.33.jar务必用--files而不是让代码里写死本地路径。另一点本地直接读 MySQL 没问题集群上如果 executor 访问 MySQL 被网络策略挡住也会出现本地成功集群失败这个要先确认安全组和连通性。5.4 现象读取JSON时schema推断结果每天都不一样现象某天跑出的read_duration变成了 string导致 sum 报异常再过一天又变成 long一个作业要改三回代码。原因spark.read.json在小文件数量多时流程是“先采样再推断”采样到的文件里恰好缺字段或类型变化就会把 schema 推断错。解决用 2.1 节的方式手动指定 schema并且打开spark.sql.json.ignoreCorruptFiles。schema 必须手工维护这也是项目文档里最该写明的第一原则。5.5 现象在Driver端collect()后内存瞬间爆掉现象分析完成后用collect()拉结果到本地大结果集直接 OOM或者集群 client 节点卡死。原因collect()会把所有分区的数据全部传到 DriverDriver 堆内存只有--driver-memory那么大结果集一旦超过它就死路一条。解决只预览数据就用take(10)或show()需要全量结果就去写 MySQL 或 HDFS不要 collect。如果确需在 Driver 做二次处理也要先repartition减小数据量并且调大--driver-memorysample_preview clean_df.repartition(1).take(10)repartition(1)不推荐用于大结果会做一次全量 shuffle但用于 take 预览可接受。日常开发时我会在 Spark UI 的 SQL 页看 Physical Plan凡是看到 Exchange 节点有大量分区再确认 shuffle 增量是否合理。6. 从跑通到跑稳用记录数核对来验证数据管道很多项目跑通一次之后第二次换了时间参数结果数字对不上最后发现是清洗时把 null 全部丢掉了。这个问题的根源不是代码 bug而是缺少验证步骤。我养成的习惯是每个作业末尾都输出“血缘指标”读进来多少条、过滤后剩多少条、聚合后多少行然后写进一张data_quality_log表下次跑批直接对比。raw_count raw_df.count() clean_count clean_df.count() detail_count daily_stats.count() quality_df spark.createDataFrame([ (2025-03-15, raw_log, raw_count, 日志原始条数), (2025-03-15, clean_log, clean_count, 清洗后条数), (2025-03-15, daily_report, detail_count, 日报结果行数) ], [dt, metric, cnt, note]) quality_df.write.jdbc( urljdbc:mysql://localhost:3306/ebook_analysis, tabledata_quality_log, modeappend, properties{ user: root, password: os.environ.get(MYSQL_PASSWORD, ), driver: com.mysql.cj.jdbc.Driver })这三个数字能说明很多问题如果clean_count比昨日同口径少了 50%大概率是数据源路径写错或清洗条件过严如果daily_report行数为 0那问题多半出在 dt 参数传递上。不要小看这几行代码它相当于给黑盒式的 Spark 作业装了一个仪表盘。在项目文档里我会把验证步骤写成三档第一档是“能跑通”即 spark-submit 提交后 exit code 为 0第二档是“能对上”即关键指标和业务方日报历史值偏差在可接受范围第三档是“能自动恢复”即失败重跑不会产生重复数据靠的是写 MySQL 时的写入模式。做完这三档基于 Spark 的电子书平台数据分析设计与实现才算真正交付而不仅仅是交一段代码。最后说一个我自己的习惯每次提交跑批任务之前先用本地小数据集把--dt参数写死跑一遍确认今天的日期、路径和分区都存在再上集群。这个动作帮我省了至少十次因为日期写错导致的无效跑批。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
