豆瓣电影爬虫与Spark数据分析可视化实战

简介：一份基于Python和Spark的豆瓣电影爬虫与数据分析可视化系统，适合毕业设计、期末大作业和课程设计场景。项目完整覆盖从网页爬虫、数据清洗、Spark批量统计到前端可视化展示的整个流程，面向想快速搭建大数据分析应用的Python和Spark初学者。资源包共包含241个文件，压缩包大小约5.65MB，主体以XML配置和Spark中间结果文件为主，同时配有Java逻辑代码、HTML/CSS/JS页面、SQL数据库脚本以及Python爬虫与Spark处理脚本，目录结构清晰便于按模块查阅。从处理输出可以看出，系统已实现影评词频、电影类型、评分等级、评论数、年份分布等多个维度的统计与可视化，能直接用于项目演示或二次开发。当前已有248人学习下载。代码注释详细，新手也能看懂，附带数据库文件，简单部署即可运行；作为作者手打98分的毕业设计，兼具完整性和实用价值，尤其适合作为课程设计、期末作业和毕设参考。

1. 为什么不直接爬完就展示：Spark在这套豆瓣电影系统里的位置

如果只是把豆瓣电影Top250爬下来存到Excel，那这项目最多算入门作业。真正的分水岭在于你能否回答"近五年哪个类型的电影评分中位数最高"、"评论里出现频率最高的形容词是哪几个"这类问题。这套基于Python+Spark的方案，爬虫只负责收集，真正的分析落在了Spark的分布式计算上。

项目里出现的WordNum.class、TypeNum.class这类文件，并不是Python源码，而是Spark或MapReduce作业编译后的class。也就是说，爬虫脚本和批处理分析代码是两套技术栈，中间通过MySQL衔接。这种混合结构恰恰是真实数据工程里最常见的样子：采集用Python，批量计算用Spark。

你拿到目录里除了Python文件，还有数据库文件(.sql)和Spark作业源码，意味着不需要从头建表抓数据，导入数据库就能开始跑完整链路。适合期末大作业和课程设计作为参考，也适合想搞清楚"爬虫+离线分析+可视化"三者怎么协作的人。

2. 爬虫设计：requests、代理池与反爬降级的取舍

2.1 目标URL与请求头设计

豆瓣电影需要抓两类页面：列表页和详情页。列表页拿到电影ID，详情页补全评分、评价人数、类型、语言这些字段。用requests而不是scrapy，主要原因是项目规模不大，requests配合Session足够控制请求频率，也能更精细地处理异常。

import requests import time from fake_useragent import UserAgent ua = UserAgent() HEADERS = { "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", } def fetch_movie_detail(movie_id: int) -> dict: url = f"https://movie.douban.com/subject/{movie_id}/" try: resp = requests.get( url, headers={"User-Agent": ua.random, **HEADERS}, timeout=3 ) resp.raise_for_status() resp.encoding = "utf-8" return parse_detail(resp.text) except requests.RequestException as e: print(f"[warn] {movie_id} failed: {e}") return {}

这里的timeout=3是为了防止一个电影详情页卡死整个爬虫线程，resp.encoding = "utf-8"强制编码，避免豆瓣页面里偶尔的编码误判。ua.random每次请求随机一个User-Agent，降低被规则拒绝的概率。解析函数parse_detail一般用BeautifulSoup按meta标签提取数据，比如：

from bs4 import BeautifulSoup def parse_detail(html: str) -> dict: soup = BeautifulSoup(html, "html.parser") info = soup.select_one("div#info") if not info: return {} return { "title": soup.select_one("h1 span").get_text(strip=True), "year": extract_year(info), "rating": extract_rating(soup), }

这段代码没有做全量字段解析，而是保留extract_year和extract_rating两个辅助函数的位置。实际做毕设时，你可以在里面分别用正则re.search(r"(\d{4})", text)提取年份，用property="v:average"取评分，比一次写死更可维护。

2.2 代理池和限速策略

豆瓣对单IP的请求频率非常敏感，连续请求超过几十次就会弹出验证码。常见做法是维护一个代理池，每次请求前从池子里随机取一个代理。这里有一个很小的调度模块：

import random PROXIES_POOL = [ "http://user:pass@proxy1:8080", "http://user:pass@proxy2:8080", ] def get_proxy(): return {"http": random.choice(PROXIES_POOL), "https": random.choice(PROXIES_POOL)} def crawl_with_retry(movie_id: int, retries: int = 3): for attempt in range(retries): resp = fetch_movie_detail(movie_id, proxies=get_proxy()) if resp: return resp time.sleep(2 + attempt * 2) return {}

重试间隔按2 + attempt * 2递增，第二次重试等4秒，第三次等6秒，尽量避开瞬时封禁。并发设计方面，不要为了赶进度直接开50个线程，否则封禁概率会指数上升。一般用ThreadPoolExecutor(max_workers=5)加Semaphore控制并发量。真正追求速度的分布式爬虫，通常会让代理池每分钟轮换几百个IP，但那是有商业代理支撑的方案，毕设里未必有必要。

2.3 数据落库的表结构

爬下来的数据最终要落到MySQL里，Spark再读库分析。表结构不能设计成一把抓，否则后面分析会很痛苦。常见设计是movie和movie_comment各一张表：

评论表至少要有movie_id、comment_text、comment_time三个字段。这里的genres字段存成逗号分隔字符串，会让后续Spark explode操作很顺手，后面第四章会看到具体用法。主键就看电影编号，如果重复爬取，用INSERT ... ON DUPLICATE KEY UPDATE更新评分和评论数，保证数据幂等。

2.4 真实环境中的反爬变化

豆瓣的反爬策略这些年变严格了很多。改User-Agent只能防住最简单的拦截，真正要稳定抓取，还得关注Cookie里的bid字段。很多爬虫第一次访问会先抓首页拿到Cookie，再把Cookie塞进Session，看起来更像真实用户。

我一般会在requests.Session里维护同一个Cookie，并把单次请求间隔调成1到2秒。如果发现页面返回登录跳转，就说明当前代理已经被标记，需要换IP。更保险的方式是用豆瓣的公开API，比如搜索接口，但那种接口也受频率限制，和爬页面没有本质区别。这个项目里的爬虫部分重点在演示链路，而不是挑战高强度反爬，所以放在本地小批量跑就完全够用。

3. Spark作业读取MySQL：从原始数据到可分析的DataFrame

3.1 为什么是Spark而不是Pandas

几万条电影记录Pandas完全能处理，但毕设的评分要求里通常会强调"分布式计算"。Spark的价值在于当评论数据膨胀到几百万条时，同样的聚合逻辑可以在多台机器上并行。更重要的是，Spark SQL可以直接对MySQL建临时视图，用纯SQL跑聚合，这对不熟悉DataFrame API的人非常友好。

另一个原因是项目中需要做中文分词统计，Pandas要自己写并行很容易写出bug，Spark的map、reduceByKey操作天然适合文本统计。如果你本机没有Spark集群，常见做法是先搭standalone模式，至少一个master一个worker，本地用小分区数模拟分布式执行。

3.2 JDBC读取与分区设置

Spark读取MySQL不是把整表抽到内存，而是通过JDBC连接器按分区读取。如果不设置分区，Spark会对全表产生一个Task，数据量大时直接OOM。设置分区字段和范围能让多个Task并行拉数据。

val jdbcDF = spark.read .format("jdbc") .option("url", "jdbc:mysql://localhost:3306/douban") .option("dbtable", "movie") .option("user", "root") .option("password", "your_password") .option("driver", "com.mysql.cj.jdbc.Driver") .option("partitionColumn", "id") .option("lowerBound", 1) .option("upperBound", 1000000) .option("numPartitions", "8") .load()

partitionColumn必须是一个数值型字段，一般用自增主键。Spark会把范围从lowerBound到upperBound切成numPartitions段，每段对应一个WHERE id >= ? AND id < ?的子查询。注意numPartitions不要大于数据库最大连接数，否则每个分区建一个连接，会把MySQL连接数打满。

这里还有几个容易踩的坑：MySQL驱动要放进Spark的jars目录，serverTimezone=Asia/Shanghai要加到url里，否则报日期时区错误。dbtable可以写成(SELECT id,title,rating,genres,comment_count FROM movie) tmp，先投影再拉取，减少网络传输。

3.3 清洗和类型转换

原始表里year可能是字符串，rating也可能包含缺失值。Spark的强类型API要求我们在计算前把这些字段转成Int和Double。下面的代码处理了三种异常：

import org.apache.spark.sql.functions._ val cleanDF = jdbcDF .filter(col("title").isNotNull) .withColumn("year", col("year").cast("int")) .withColumn("rating", col("rating").cast("double")) .withColumn("comment_count", col("comment_count").cast("int")) .na.fill(Map("rating" -> 0.0, "comment_count" -> 0))

cast("int")会把"2009"这种字符串转成数字，但也会把"None"转成null。所以先用isNotNull过滤掉关键字段为空的记录。na.fill用来补评分和评论数为缺失的记录，后面按评分排序时就不会出现null跑到最前面。这个顺序很关键：先过滤，再转换，最后填充，如果顺序错了，填充完的类型又会被cast成null。

3.4 中文分词与WordUtil

项目中出现的WordUtil.class，它的作用是把评论按中文分词拆开，再统计每个词的出现次数。如果你用的是Spark，常见实现是引入HanLP或者结巴分词。但HanLP的词典文件比较大，在Spark集群上每个executor都加载一次，会给内存增加负担。

常见的做法是把分词封装成一个map函数，让Spark对每个分区执行：

val seg = new Segment() val commentDF = spark.read.table("comment").select("comment_text") commentDF.map(row => { val text = row.getString(0) val words = seg.segment(text).asScala.map(_.word.trim).filter(_.length > 1).toList (words, 1) }).rdd.flatMap(pair => pair._1.map(word => (word, pair._2))) .reduceByKey(_ + _)

这里的_ .map和reduceByKey就是RDD里最常见的wordcount逻辑。有一个细节是分词词典需要把"电影"这类停用词过滤掉，否则统计结果全是"的"、"是"这种无意义词。我一般会加载一个停用词表，在filter的时候顺手排除。还有一次我在集群上跑，发现输出里大量乱码，后来确定是新节点没有安装系统级别的中文字体，Spark executor打印日志时编码异常，这不是业务bug，而是环境问题。

4. 指标计算：评分分布、类型热度和年份趋势的实现

4.1 评分区间统计

评分分布通常是按"9分以上、8-9、7-8、6-7、6以下"分组，统计每组电影数量。这种分桶逻辑用Spark SQL最清晰：

cleanDF.createOrReplaceTempView("movie") spark.sql(""" SELECT CASE WHEN rating >= 9 THEN '9+' WHEN rating >= 8 THEN '8-9' WHEN rating >= 7 THEN '7-8' WHEN rating >= 6 THEN '6-7' ELSE '6-' END AS rating_bucket, COUNT(*) AS cnt FROM movie GROUP BY rating_bucket ORDER BY rating_bucket """)

CASE WHEN的求值顺序是从上到下，所以每个区间的下界是前一个区间的上界，不需要再写rating < 9 AND rating >= 8这种冗余条件。GROUP BY rating_bucket其实是对表达式分组，但SQL里可以直接用别名，Spark SQL支持这个语法。

4.2 类型榜单分析

genres字段存的是"剧情, 爱情, 战争"这种逗号分隔字符串，要做类型榜，必须先把每个电影按类型拆开。这里需要用split配合explode：

val typeDF = cleanDF .withColumn("type", explode(split(col("genres"), ","))) .groupBy("type") .agg( count("*").alias("movie_count"), avg("rating").alias("avg_rating") ) .orderBy(desc("movie_count"))

split(col("genres"), ",")把字符串转成数组，explode再把数组的一行拆成多行。关键在于，如果genres为空，explode会直接把这行数据丢掉，而实际业务上确实有电影没标注类型。想保留这些电影可以换成explode_outer。类型名的空白也需要处理，我常在split前用regexp_replace(col("genres"), "\\s+", "")去掉所有空格，否则每个类型前面会带空格，导致统计不准确。

4.3 年份维度聚合

年份趋势适合看的是"每年平均评分"或者"每年电影数量"。年份字段是int后，直接groupBy("year")即可。比较实用的一个做法是把年份按十年分桶，看长周期趋势：

spark.sql(""" SELECT CONCAT(FLOOR(year/10)*10, 's') AS decade, ROUND(AVG(rating), 2) AS avg_rating, COUNT(*) AS movie_count FROM movie WHERE year >= 1970 GROUP BY FLOOR(year/10) ORDER BY decade """)

这里用FLOOR(year/10)*10得到1980、1990这种整十年起点，再用CONCAT拼一个"s"后缀。ROUND(AVG(rating), 2)控制小数位数，因为评分均值经常出现9.200000这样的长尾。不加WHERE year >= 1970的话，一些年份为0的脏数据会单独成组，整个趋势图最左边会莫名其妙多一列。

4.4 评论情感与词频

评论数据量大，直接用完整文本做情感分析较慢。毕设场景里更常见的是统计评论中的高频词，用词频变化代表关注度。我们可以在分词的基础上，把每个词的频次和它出现的影评关联，形成评论关键词表。

这里有一个很关键的Spark内存问题：reduceByKey会在shuffle前先做本地的merge，所以不会把全量数据放在一个executor里。但如果中间出现了很大的key列表，比如几万个电影id，每组的评论数巨大，就要考虑提高spark.default.parallelism和executor内存。

通常我在spark-submit里这样设置：

spark-submit \ --master local[4] \ --executor-memory 2g \ --conf spark.sql.shuffle.partitions=10 \ --conf spark.default.parallelism=10 \

spark.sql.shuffle.partitions决定shuffle后的分区数，默认200，在小数据集上会产生非常多空任务，浪费调度开销。调成10可以明显减少CPU空转。如果你的集群内存只有4G，executor-memory不要超过3g，给系统留一点余量。

5. 可视化：ECharts大屏怎么接Spark结果

5.1 后端接口设计

Spark算出的结果最终要落到MySQL或导出成JSON，可视化层不能直接读MySQL大表。常见做法是让Flask提供一个统计接口，接口内部只做读库和返回JSON，具体SQL已经在Spark端算好，前端拿到的就是聚合后的数据。

from flask import Flask, jsonify import MySQLdb app = Flask(__name__) db = MySQLdb.connect(host="localhost", user="root", passwd="123456", db="douban", charset="utf8mb4") @app.route("/api/rating_bucket") def rating_bucket(): cursor = db.cursor() cursor.execute("SELECT rating_bucket, cnt FROM rating_stat ORDER BY rating_bucket") rows = cursor.fetchall() return jsonify({"categories": [r[0] for r in rows], "data": [r[1] for r in rows]}) if __name__ == "__main__": app.run(port=5000)

这里charset="utf8mb4"必须显式声明，否则中文在前端很容易变成问号。接口返回的categories和data分离，前端ECharts可以直接映射到x轴和y轴，不需要再做二次处理。注意如果统计数据是Spark离线算好后写回一张rating_stat表，那么这份接口代码里查询的表名就是它。

5.2 前端的图表配置

ECharts里最常用的三种图是柱状图、饼图和折线图。柱状图适合评分区间，饼图适合类型占比，折线图适合年份趋势。下面是一个典型的柱状图配置：

fetch('/api/rating_bucket') .then(response => response.json()) .then(res => { const chart = echarts.init(document.getElementById('ratingChart')); chart.setOption({ title: { text: '豆瓣电影评分区间分布' }, tooltip: {}, xAxis: { type: 'category', data: res.categories }, yAxis: { type: 'value' }, series: [{ type: 'bar', data: res.data, itemStyle: { color: '#5470c6' } }] }); });

fetch是异步请求，ECharts必须在数据返回后再init，否则表格的宽高是0，图会画不出来。category类型的xAxis要求数据是字符串数组，如果后端返回数字，还需要String()转换一次。这类细节往往是毕设答辩现场报错的原因。

5.3 从静态JSON到动态刷新

如果只是做一个只读大屏，把数据写死在JavaScript里也能交差。但更贴近真实项目的做法是，让前端每隔一段时间重新拉一次接口。

function loadData() { fetch('/api/rating_bucket') .then(r => r.json()) .then(res => chartRef.current.setOption(/* ... */)); } setInterval(loadData, 60 * 1000);

setInterval(loadData, 60 * 1000)代表每分钟刷新一次。这样做的好处是，Spark每天跑一次批处理生成新统计结果，大屏跟着自动变化，不需要手动刷新页面。但要注意时间间隔不要太短，否则后端会被反复查询压力打满。配合setInterval，在组件卸载时一定要clearInterval，否则浏览器内存会一直挂着定时器。

5.4 可视化大屏适配注意事项

用过可视化大屏的人都知道，分辨率不同布局会乱。最稳妥的方案是使用rem方案，把设计稿宽度设为1920px，前端所有尺寸按比例换算。ECharts的图表容器用width: 100%; height: 100%，再由父容器控制宽高。

另一个细节是ECharts图表的grid配置，尤其是饼图和折线图。很多新手把图放在一个Flex容器里，发现图被拉伸变形。遇到这种情况，先检查echarts.init时容器是否已经可见、有没有固定宽高。还有一个更容易忽略的点，是如果页面里有多个图表，必须为每个图表的div分配独立的ref或id，不能让多个chart实例共享同一个节点。

6. 部署与验证：用自带数据库文件快速跑通，别在Spark环境上卡住

6.1 环境准备

这个项目给你省了不少事，数据库文件已经准备好，不需要自己建表。但环境还是得装齐。

# Python 依赖 pip install requests beautifulsoup4 flask pymysql # Spark 环境 docker pull bitnami/spark:3.5

Python爬虫和可视化基本不依赖Spark本机安装，真正需要Spark的只有分析模块。如果不想手动搭集群，用Docker跑一个spark standalone是最快的路径。记住docker pull之后要把spark的bin目录挂进容器，或者直接用容器内的spark-submit命令执行。

6.2 数据库导入

项目自带的数据库文件通常是.sql，导入MySQL：

mysql -uroot -p -e "CREATE DATABASE douban DEFAULT CHARACTER SET utf8mb4" mysql -uroot -p douban < douban.sql

DEFAULT CHARACTER SET utf8mb4非常重要，如果创建库的时候用了默认latin1，后面中文数据插入时会报编码错误。导入完成后先执行一条SELECT COUNT(*) FROM movie;，看数量是否与文档一致，能提前发现.sql文件被截断的问题。

6.3 最小复现流程

要把整个项目跑起来，建议按这个顺序操作：

1. 导入数据库后，先运行爬虫脚本抓几条新数据，验证数据库能写入。
2. 运行Spark分析脚本，把统计结果写入rating_stat等结果表。
3. 启动Flask后端，访问/api/rating_bucket确认返回JSON。
4. 打开前端页面，看图表是否正常渲染。

不要让爬虫大规模运行，因为服务器早已不是当年文档里的环境，一夜之间把IP封了很正常。先limit 20条跑通链路，再决定是否扩展。

6.4 如何查看Spark Job执行情况

最后一个实用技巧，是学会看Spark的监控页面。无论本地local模式还是standalone集群，提交任务后都会在http://localhost:4040暴露一个Web UI。

docker exec -it spark-master spark-submit --master local[2] /opt/analyse.jar

打开localhost:4040/jobs/，可以看到每个Stage的输入记录数和耗时。如果某个Stage的Input远远大于预期，说明读取MySQL时没有限制分区字段，全表扫描了。根据这个信息，可以反向调整JDBC的lowerBound和upperBound，而不是盲目调大executor内存。这是整个项目里最值得花时间研究的地方，理解了Spark的任务分配，才算真的把爬虫之后的计算链路吃透。

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