Hadoop实战：基于MapReduce的豆瓣电影数据分析与可视化

简介：Hadoop豆瓣电影分析可视化源码是一套面向大数据课程实验的完整项目参考，围绕豆瓣电影Top250榜单数据，模拟真实大数据分析场景，适用于本科及高职大数据专业的课程设计、Hive案例实践与毕业设计参考。项目需要搭建Hadoop集群，集成HDFS、HBase、Hive、Flume、Sqoop等组件，将豆瓣文本或CSV数据导入Hive完成预处理和统计分析，再通过Python与ECharts实现可视化展示。压缩包仅14KB，共10个文件，由5个Python脚本和5个HTML页面组成，脚本主要完成数据爬取、评分统计、国家走势追踪、综合评分Top10分析等任务，HTML则呈现类型时长占比、年代人气度、剧情电影评分趋势等交互图表。目前已有6305人学习下载。借助这套源码，可理清从数据采集、存储、分析到可视化的完整大数据链路，参考实际代码与图表逻辑，快速复现实验或进行二次开发，是完成Hadoop课程设计的实用参考。

1. 豆瓣电影这套分析，为什么落到 Hadoop 上才值得写

几乎每个做过爬虫的人都写过“豆瓣电影 Top250 统计”这类练手项目，但绝大多数止步于把数据存进 CSV、再用 pandas 画两张图。换到 Hadoop 上之后，问题的性质变了：你不再只是处理 250 条数据，而是把整套流程前置到分布式存储与计算模型之下，学习 MapReduce 的“状态化计算”到底怎么设计。很多人把 Hadoop 当成一个“大数据版的 Excel”，这是误解。Hadoop 的价值不在单次计算有多快，而在它允许你要先把数据切到 HDFS 上、再按批处理范式去算，运算逻辑一旦写好就可以横向扩到更大的数据集。这篇文章就以“分析豆瓣电影数据并做可视化”这条链路为主线，讲清楚从数据落地、MapReduce 作业设计、到给前端提供 JSON 数据服务的完整实现路径。适合正在做课程设计、或者准备 Hadoop 面试想拿一个能讲深讲透的实战项目的人。

2. 豆瓣数据采集与 HDFS 目录设计

2.1 爬虫数据侧：字段约定比爬多少条更重要

做分析项目第一步不是写爬虫，而是先定字段契约。豆瓣电影详情页能拿到的字段很多，但对一个 Hadoop 分析项目来说，有用的通常只有这几类：电影ID、片名、评分、评分人数、年份、制片国家、类型（一个电影有多个类型）、导演和主演（逗号分隔）。最终落成 CSV 时，我建议统一用\t做列分隔符，而不是逗号——因为电影类型和演员字段里大量出现逗号，再用逗号做分隔符清洗起来非常痛苦。

import requests from bs4 import BeautifulSoup import csv import time import random headers = { "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36" } def fetch_detail(douban_id): url = f"https://movie.douban.com/subject/{douban_id}/" resp = requests.get(url, headers=headers, timeout=10) soup = BeautifulSoup(resp.text, "html.parser") # 此处按业务需求提取评分、片名、年份等字段，注意用 try 包裹避免单条解析失败中断任务 ... with open("douban_movie.tsv", "w", encoding="utf-8-sig") as f: writer = csv.writer(f, delimiter="\t") writer.writerow(["movie_id", "title", "rating", "year", "genres", "region", "director"]) for mid in id_list: data = fetch_detail(mid) if data: writer.writerow([mid] + list(data.values())) time.sleep(random.uniform(1, 3))

这里有一个已经决定后面计算逻辑的关键点：genres 是多值字段。比如《霸王别姬》的类型是“剧情/爱情”，写入 CSV 时我推荐用/拼接成一个字符串，这样在 Map 阶段拆开的时候不会和列分隔符冲突。random.uniform(1, 3)的 sleep 是必须的，不是形式主义——要抓全量豆瓣样本，没有合理的请求间隔会导致 IP 被临时限制，后面补数据比多等两秒更耗时。字段提取部分用try包住单条解析逻辑，避免因为一部电影的页面结构异常而中断整个采集任务。

2.2 HDFS 上的目录布局：让分析作业少写过滤条件

把数据放进 HDFS 时，绝大多数人会随手扔到/douban/input/movie.tsv。当一个项目只有一张表时这么设计没问题，但你的分析目录马上会膨胀：原始数据、清洗后数据、MapReduce 输出、可视化服务读取的数据。我习惯在项目一开始就建好分模块目录：

hdfs dfs -mkdir -p /douban/raw hdfs dfs -mkdir -p /douban/cleaned hdfs dfs -mkdir -p /douban/analysis/rating_dist hdfs dfs -mkdir -p /douban/analysis/year_trend hdfs dfs -mkdir -p /douban/analysis/genre_rating hdfs dfs -mkdir -p /douban/export

这个分区的意义在后面会体现出来：每个 MapReduce 作业的输入输出路径固定到具体子目录，作业之间不会因为互相依赖而产生“重复过滤同一批数据”的代码。分析作业直接读/douban/analysis下的结果目录，而不是每次从头读全量原始表。伪分布式环境只有几个节点，你在写hdfs dfs -put之前就把目录建好，比事后用-mv挪数据省很多事。

hdfs dfs -put ./douban_movie.tsv /douban/raw/douban_movie_202406.tsv hdfs dfs -ls /douban/raw/

2.3 伪分布式环境的集群参数雏形

如果读者所在环境是单机伪分布式，有一点必须先确认：默认副本数是 3，但伪分布式只有一个 DataNode，HDFS 会一直报副本不足的告警。虽然不影响分析作业执行，但日志里刷错会干扰你排查问题。在hdfs-site.xml里显式配置副本数：

<property> <name>dfs.replication</name> <value>1</value> </property>

本机开发环境把副本数降为 1，集群资源消耗小，跑测试作业时也不会因为等待副本写入而拖慢速度。但这里要打一个预防针：这只是开发配置，真正的多节点集群不要改这个值，保持默认 3 才能保证节点故障时数据不丢。面试被问到“HDFS 副本策略”时，能说出这个改动背后的权衡，比背十遍默认值管用。

3. MapReduce 分析任务设计与参数调优

3.1 评分分布统计：从单 JOB 讲透 Map 和 Reduce 的边界

先从一个最简单但能覆盖完整链路的作业开始：统计不同评分区间的电影数量。评分是 1~10 的小数，我们按整数部分分组（9.1 分归到 9 分组）。这个作业用来理解 Mapper、Reducer 和 Partitioner 各自承担什么职责。

import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; public class RatingDist { public static class RatingMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable> { private static final IntWritable ONE = new IntWritable(1); private IntWritable ratingGroup = new IntWritable(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\t"); // 约定：第 3 列是评分。脏数据或表头必须在这里过滤掉 if (fields.length < 3 || fields[2].equals("rating")) { return; } try { double rating = Double.parseDouble(fields[2]); ratingGroup.set((int) Math.floor(rating)); context.write(ratingGroup, ONE); } catch (NumberFormatException e) { // 评分列解析失败时不中断作业，统计到自定义计数器里供后续排查 context.getCounter("Parse", "InvalidRating").increment(1); } } } public static class RatingReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> { private IntWritable result = new IntWritable(); @Override protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "douban rating distribution"); job.setJarByClass(RatingDist.class); job.setMapperClass(RatingMapper.class); job.setCombinerClass(RatingReducer.class); job.setReducerClass(RatingReducer.class); job.setOutputKeyClass(IntWritable.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

这个作业里有一点需要特别解释：setCombinerClass(RatingReducer.class)。Combiner 是 Map 端的本地汇总，它可以在 map 输出落盘之前把同一个 key 的重复值先合并，减少传给 reducer 的数据量。这里 Combiner 和 Reducer 用同一个类，因为我们的操作是求和，满足交换律和结合律。但如果换成计算平均值，这个设计就是错的——Combiner 合并了平均值之后，Reducer 拿到的是“平均值的平均值”，和真正想要的结果完全不同。

评分分布这个例子对应了 Hadoop 面试里很高频的一个问题：哪些操作可以用 Combiner，哪些不能。计数、求和、最大值都能用；平均值、中位数、去重计数不能用。

3.2 年度评分趋势：两个 JOB 串联时，中间结果怎么定

年度评分趋势要计算的是“每一年上映电影的平均评分”。在 MapReduce 里没有 SQL 那种GROUP BY的临时表概念，每个作业一次只输出一层粒度。这里必须用两次 MapReduce：第一个作业把“年份、评分”作为输出，第二个作业按年份聚合计算平均值。

public class YearRatingStep1 { // Mapper 输出 key: 年份, value: 原始评分 public static class YearMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> { private Text yearKey = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\t"); if (fields.length < 4 || fields[3].equals("year")) { return; } yearKey.set(fields[3]); context.write(yearKey, new DoubleWritable(Double.parseDouble(fields[2]))); } } // Reducer 输出 key: 年份 带一个特殊标记, value: 总和和条数的拼接 }

第二步的 Reducer 需要的不仅是总和，还需要知道有多少条数据参与计算，所以第一步的输出值必须携带总数，写成总和:条数的拼接字符串。这里有几个常见的坏味道值得避开：不要在第一个作业里就计算平均值，然后让第二个作业对“平均值”再做平均，因为每年的电影数量权重不同，这样算出来的整体趋势是被扭曲的。

实际生产里，两个作业串联时中间结果应该放在独立的临时目录里，不要在第二个作业的输入参数里直接指向第一个作业的输出目录再要求它整改——HDFS 上不允许对正在写入的目录做重名覆盖。常见的做法是第一个作业写到/douban/analysis/year_sum_tmp，第二个作业读取该目录后写到/douban/analysis/year_trend，全部跑完后手动执行hdfs dfs -rm -r清理临时目录，或者直接写一个 Shell 脚本把两段提交串起来。

hadoop jar douban-analysis.jar YearRatingStep1 /douban/raw/douban_movie_202406.tsv /douban/analysis/year_sum_tmp hadoop jar douban-analysis.jar YearRatingStep2 /douban/analysis/year_sum_tmp /douban/analysis/year_trend hadoop fs -rm -r /douban/analysis/year_sum_tmp

用命令行串联两个作业够用，但如果作业数量增长到四五个，手动执行很容易出错或漏跑。更可控的方式是用 Oozie，或者直接在客户端代码里用JobControl建立依赖有向无环图，这里不展开，但需要知道这种演进路径的存在。

3.3 类型维度的自定义 Writable：别把多值字段压平了

类型统计比前两个场景复杂：一部电影有多个类型，如果直接解析genres字段后在 Map 里按/切分再输出，一部电影会被分发到多个类型对应的 Reducer 中。这样算“每个类型的平均评分”时，同一部电影会重复计入多个类型。在某些业务场景这没问题——比如分析“喜剧片整体评分”是合理的。但如果你想统计的是“所有电影在类型维度的分布”，重复计数就会让你的饼图加起来超过 100%。

自定义 Writable 的用武之地在这里：你需要在map阶段输出一个复合 key，key 既要携带类型，还要保留原始电影 ID，Reducer 端用电影 ID 去重。

import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class MovieTypeKey implements WritableComparable<MovieTypeKey> { private Text type = new Text(); private Text movieId = new Text(); public MovieTypeKey() {} public MovieTypeKey(String type, String movieId) { this.type.set(type); this.movieId.set(movieId); } @Override public void write(DataOutput out) throws IOException { type.write(out); movieId.write(out); } @Override public void readFields(DataInput in) throws IOException { type.readFields(in); movieId.readFields(in); } @Override public int compareTo(MovieTypeKey other) { int cmp = type.compareTo(other.type); if (cmp != 0) { return cmp; } return movieId.compareTo(other.movieId); } }

构造这个复合 key 之后，Reducer 收到的数据按类型分组，组内按 movieId 排序，Reducer 里对同一个 movieId 只计一次。排序由 Hadoop 的 Shuffle 机制自动保证。这里真正值得理解的是：WritableComparable接口要求你同时实现write、readFields和compareTo，前两个管序列化和反序列化，compareTo决定 Shuffle 阶段的排序规则。如果你不实现自定义 key，而是塞一个拼接字符串，那么 Reducer 端就必须自己维护去重状态，编程复杂度会显著增加，而且容易出错。

3.4 参数调节：从默认值到能应对亿级数据

针对豆瓣电影这个量级，真正值得调的不是内存参数，而是把mapreduce.job.reduces设置成和你的目标输出文件数量一致。比如最终要导出到可视化服务的聚合结果，只有几百行的量级，那么 reduce 数量设个 1~2 个就好，文件多了前端还要合并。如果以后数据量扩大，再考虑每个 reducer 的输出采用MultiOutputFormat分目录写入。

调整参数的方式是在main里写进Configuration，或者在提交作业时加-D前缀：

hadoop jar douban-analysis.jar RatingDist -D mapreduce.map.memory.mb=2048 -D mapreduce.map.output.compress=true /douban/raw /douban/analysis/rating_dist_v2

用-D的好处是参数跟着命令走，不会污染代码库里的默认配置；缺点是每次都要手打，容易漏参数。项目里有多个作业时，我倾向于把基础参数写进mapred-site.xml，占了全局，个别作业的特殊参数再在命令行覆写。

4. 从 HDFS 到可视化大屏的数据服务

4.1 为什么聚合结果要导回关系型数据库

MapReduce 计算结果落在了 HDFS 的part-r-00000文件里，但 Web 端可视化服务不可能直接去读 HDFS 上的文件——HDFS 的定位是批处理存储，不是低延迟查询。这里常规的做法是写一个简单的导出程序，把聚合结果从 HDFS 拉下来，导入 MySQL。没必要为了这个量级引入 Hive 或 Impala，反而增加运维负担。

hdfs dfs -cat /douban/analysis/year_trend/part-r-00000 | while read line; do year=$(echo "$line" | awk -F '\t' '{print $1}') avg=$(echo "$line" | awk -F '\t' '{print $2}') mysql -u root -p123456 douban_db -e "INSERT INTO year_trend (year, avg_rating) VALUES ('$year', '$avg');" done

这段 Shell 直接在命令行中逐行解析 result 文件并写进 MySQL，适合一次性的数据初始化。但要注意这是“暴力导入”，如果avg字段含有引号或特殊字符，会导致 SQL 语句结构被破坏。这是线上环境必须避免的，更稳的做法是用一个 Java 或 Python 脚本调用数据库驱动做批处理插入，代码里用占位符转义。

MySQL 建表要按维度拆开，不要把所有统计结果都堆到一张大宽表里。年度趋势单独一张表，类型分布单独一张表，评分分布单独一张表，这样后面接可视化接口时，一个接口对应一张表，逻辑清晰。

CREATE TABLE rating_dist ( rating_group INT PRIMARY KEY, movie_count INT NOT NULL ); CREATE TABLE year_trend ( year VARCHAR(8) PRIMARY KEY, avg_rating DECIMAL(3, 1) NOT NULL ); CREATE TABLE genre_rating ( genre VARCHAR(20) PRIMARY KEY, avg_rating DECIMAL(3, 1) NOT NULL, movie_count INT NOT NULL );

4.2 Flask 提供 JSON 接口给可视化前端

后端接口不需要太重的框架，一个 Flask 应用、两个接口足够。如果你已经熟悉 Spring Boot，用它也一样；对于课程设计和快速验证来说，Flask 更短平快。

from flask import Flask, jsonify import pymysql app = Flask(__name__) def get_conn(): return pymysql.connect( host="localhost", user="root", password="123456", database="douban_db", charset="utf8mb4" ) @app.route("/api/rating_dist") def rating_dist(): conn = get_conn() cur = conn.cursor(pymysql.cursors.DictCursor) cur.execute("SELECT rating_group AS name, movie_count AS value FROM rating_dist ORDER BY rating_group") data = cur.fetchall() cur.close() conn.close() return jsonify({"code": 0, "data": data}) @app.route("/api/year_trend") def year_trend(): conn = get_conn() cur = conn.cursor(pymysql.cursors.DictCursor) cur.execute("SELECT year, avg_rating FROM year_trend ORDER BY year") data = cur.fetchall() cur.close() conn.close() return jsonify({"code": 0, "data": data}) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000)

jsonify返回的字段名要跟前端约定好。这里我统一用的是name和value这组键名，原因是 ECharts 的饼图和柱状图系列的默认数据格式就是{ name: ..., value: ... }，后端直接对齐前端的数据格式，少一层转换逻辑。注意每次查询后都要显式cur.close()和conn.close()，不然连接数会被耗尽，服务跑一晚上之后开始报Too many connections错误。

4.3 ECharts 配置与可视化大屏常见显示坑

前端页面如果只展示单张图表，直接一个 HTML 文件引用 ECharts 的 CDN 就够。做大屏项目时，三个以上图表在同一个页面里，要注意 ECharts 实例的销毁重建问题，尤其当页面需要根据筛选条件重新请求数据并刷新图表时。

async function loadYearTrend() { const resp = await fetch("/api/year_trend"); const json = await resp.json(); const chart = echarts.init(document.getElementById("yearTrendChart")); chart.setOption({ tooltip: { trigger: "axis" }, xAxis: { type: "category", data: json.data.map(d => d.year) }, yAxis: { type: "value", name: "平均评分" }, series: [{ type: "line", data: json.data.map(d => d.avg_rating), smooth: true, areaStyle: { opacity: 0.15 } }] }); }

echarts.init绑定到已经存在的 DOM 节点上。如果同一个容器被初始化两次，控制台会抛一个警告，页面上的图表只更新最后一个实例。很多人第一次接大屏都会遇到这个问题，处理方法是每次加载数据前先调用echarts.dispose销毁原来的实例再重建，或者直接用chart.clear()配合setOption覆盖。在图表数量多、切换频繁的大屏项目里，正确管理实例生命周期是主要的维护负担。年份字段注意用字符串类型而不是数字，避免早年数据（如1920）被JSON.parse自动转成数值后丢失前导零或精度变化。

5. 伪分布式环境下三招定位数据倾斜

数据倾斜是很多准备面试的人“背过但没实际处理过”的问题。其实在单机伪分布式环境下，完全可以主动制造并诊断一次数据倾斜。最经典的触发方式是在类型统计里把“纪录片”这种高产量类型的数据人为加重，或者故意在 Mapper 输出的 key 里掺杂大量空值。

第一招：观察 reducer 的耗时分布。跑完作业后，在 ResourceManager 的 Web 界面（或运行日志中）看每个 reduce task 的处理时间。正常分布是几个 task 耗时接近，如果一个 task 耗时明显超过其他所有 task 的总和，说明 key 的分布不均衡，数据倾斜已经形成。这里有个前提：你的 job 配了多个 reducer，如果只配了 1 个，那“倾斜”被掩盖了。

第二招：用计数器定位脏数据来源。在 Mapper 里，对空类型字段单独加一个计数器，MR 作业跑完会在控制台打印出维度：

context.getCounter("DataQuality", "EmptyGenre").increment(1);

运行结果里如果EmptyGenre出现几百次，再回头看原始数据的 genres 字段的拼接格式，通常能找到问题——比如行末多了空格、某个类型后面跟着换行符。清洗逻辑里加一个.trim()就能彻底解决，比在 Reducer 里做各种判断要干净得多。

第三招：把疑似倾斜的 key 单独提出来验证。做一个只有 Mapper 的小作业，输出每个 key 的数量分布，然后手动sort | uniq -c排序，观察最大 key 与平均值的差异。如果是空值导致的，直接在 Map 阶段过滤掉即可：

if (genre == null || genre.trim().isEmpty()) { context.getCounter("DataQuality", "FilteredEmptyGenre").increment(1); return; }

过滤空值不算投机取巧，因为空类型那条数据本身就无法参与类型维度的分析，留着它进 Reducer 只会拖慢作业。用好这三招，你可以在只有一台机器的前提下，把数据倾斜的完整链路亲自动手跑一遍——从制造问题、观察现象到定位修复。面试时被问到这个问题，能有理有据地把这个实操流程讲出来，比只背“加盐、两阶段聚合”的结论更让人信服。对豆瓣电影项目来说，补齐这一环，你手里这套源码就算真正跑通了从采集到呈现的每一个环节。

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