Hadoop+Spark+Python淘宝商品数据分析可视化系统设计与实现

如果你的毕业设计题目是“基于Hadoop、Spark、Python的淘宝商品数据分析与商品销售数据分析可视化系统”，先恭喜你：这个题目下限很低，上限也很低。它最稳妥的完成路径，是把Hadoop当存储、Spark当计算引擎、Python当ETL和接口层、ECharts当前端展示，串成一条完整链路。我帮A同学做过一套几乎一模一样的，从零搭到答辩通过，中间踩了不少坑，这次把整个过程从架构设计到实操代码、再到问题排查和答辩话术，一次讲清楚。

很多同学拿到这个题目的第一反应是“我要爬淘宝”。我的建议是：别把核心精力放在爬虫上。淘宝的反爬体系不是一朝一夕能突破的，而且毕设答辩时导师更关心的是你能不能解释清楚数据链路、有没有工程能力，而不是你有没有真的扒到几万条宝贝数据。数据可以用公开数据集，也可以自己写程序生成一套逻辑自洽的模拟数据。模拟数据只要字段合理、分布真实、口径统一，完全够撑起整条链路。

1. 题目拆解：Hadoop、Spark、Python到底各干什么活

1.1 毕业设计在考察什么

这个题目表面上是“做一个数据分析系统”，实际上考察四件事：数据采集能力、数据存储与计算能力、数据建模与指标分析能力、结果可视化与展示能力。说穿了就是学校标准的“全链路数仓Demo”：采集 → 存储 → 计算 → 可视化。你不要把它想成一个真正能上线的商业系统，它是一个教学性质的项目，所有组件存在的意义都是为了证明你掌握了这条链路上的主流工具。

所以系统至少要包含几大模块：数据采集模块（Python脚本生成/采集数据放入HDFS）、存储模块（HDFS存储原始数据，MySQL存储结果数据）、计算模块（Spark进行ETL清洗、聚合统计）、可视化模块（Flask提供JSON接口，ECharts渲染图表）。

1.2 三个技术点的角色定位

先搞清楚名字和职责。

Hadoop在这个项目里的核心是HDFS，也就是分布式文件系统。你的原始订单数据、商品数据，最终都要落到HDFS上。至于Hadoop里的MapReduce，我建议直接放弃——能用Spark做离线批处理，为什么还要用MapReduce写一坨WordCount式的代码去聚合统计？答辩时导师如果问“为什么没用MapReduce”，你完全可以回答：MapReduce需要手写更多代码来优化DAG，Spark基于内存计算，写起来清晰、调试方便、性能更好，所以选Spark作为核心计算引擎。

Spark做的是清洗、过滤、聚合。比如订单表里有空值、有脏数据、有负数的价格，你在Spark里要处理掉，然后按天、按类目、按地域、按价格区间做分组统计，最后把统计结果写到MySQL。

Python的角色分两块：一是写采集/生成脚本，产出CSV或JSON格式的原始数据；二是写可视化后端的Web服务，用Flask这种轻量框架读取MySQL结果，输出JSON给前端。注意Python不要掺和到大数据计算里，它不是用来处理几千万行订单的，那是Spark的事。

1.3 用一个类比说清楚数据流

可以把整套系统想象成一家连锁餐厅：淘宝的每一笔订单就像菜市场每天送来的食材，HDFS是你租的大型冷库，食材先塞进冷库存着；Spark是中央厨房的流水线，冷库存的食材拉到流水线上分拣、洗净、切配；MySQL是配菜间里的一排小保鲜柜，中央厨房加工好的半成品放进小保鲜柜；Flask是传菜员，把配菜间的半成品端到前厅；ECharts是菜单和桌牌，顾客看了就一目了然。

这个类比请在答辩开场时直接甩出来。导师听过的技术堆砌太多了，一个清晰的比喻能让他立刻确认你理解整条链路，而不是只会照着教程敲命令。

2. 技术选型：版本和组件怎么定才不折腾

2.1 版本搭配

版本不兼容是新手最容易翻车的地方，请直接抄下面的组合。

这个表是踩过无数坑总结出来的。我见过有人装Hadoop 2.7配JDK17，结果起不来NN进程；也见过用Python 3.11跑PySpark，直接报Python version 3.11 is not supported，都是白费时间。

2.2 集群规模怎么定

不要一上来搞三台虚拟机做集群。如果本机内存不够16G，你光启动进程就要占掉大半壁江山。我的推荐方案是：一台Linux虚拟机，内存给8-10G，部署伪分布式Hadoop，以YARN模式启动Spark。伪分布式就能验证整个HDFS + YARN + Spark链路，做毕业设计完全够用。

如果你非要集群，最低配是三台2核4G虚拟机，需要先配好免密SSH，否则每启一个进程都要输密码，心态会直接崩掉。

2.3 为什么用YARN模式而不是Spark Standalone

Spark有两种常见运行方式：Standalone和YARN。我强烈建议在YARN上跑。原因很简单：你的题目里点名了Hadoop，如果你用Standalone模式，相当于跳过YARN资源调度，答辩时导师问“YARN在哪”，你确实没用过，会显得项目“名不副实”。

YARN模式下，Spark应用的资源是向ResourceManager申请的，能从资源管理层面体现Hadoop生态的完整性。配置也简单，spark-submit时加上--master yarn就行。

2.4 数据来源：爬虫、公开数据集、还是模拟数据

淘宝反爬有多猛，不用多说。API接口要签名、Cookie要加密、Selenium一开就被识别，真不是你毕设阶段能啃下来的。我建议按以下优先级选数据源：

1. 公开电商数据集：一些竞赛平台会有脱敏后的电商订单数据，字段比较齐全，量级从几十万到几百万条都有，拿来直接用，真实性没得挑。
2. 自建模拟数据：用Python写一个生成器，按真实业务逻辑生成一年的商品、订单数据。重点是字段要合理、统计口径要前后一致。
3. 自建简易爬虫（备选）：爬一个没有强反爬的商品列表页作为补充，不要死磕淘宝。

我的项目最终选择了“模拟数据为主，公开数据集补充验证”，在论文里老老实实写明是用程序生成的模拟数据，并说明这是为了验证大数据平台链路的可行性。导师完全认可，因为毕设考察的是架构能力，不是数据真实性。数据生成脚本写得逻辑严密，反而成了加分项。

3. 数据模型和指标体系：别一上来就写代码

3.1 核心表设计

拿到题目别急着跑环境，先画数据表。以下是我最终使用的表结构。

商品表ods_products：

订单表ods_orders：

这里有一个容易被忽略的设计：订单表必须记实付金额pay_price，而不是用商品表的price去乘数量。因为商品会有促销折扣，如果你在分析口径里混用两种价格，后面客单价、销售总额全都会不准。这个细节在答辩时主动讲，导师会觉得你懂业务。

3.2 指标口径要统一

淘宝数据分析里最常算的指标不多，但口径必须想清楚：

• 销售额GMV：所有订单实付金额求和。注意是“实付”，不是“原价”。
• 销量：所有订单购买件数求和。
• 订单数：去重后的订单编号数量。
• 客单价：GMV除以订单数，而不是除以用户数。
• 类目销售占比：某类目GMV除以全量GMV。
• 价格区间分布：按商品单价划档，0-50、50-100、100-200、200-500、500以上，算出各区间订单量占比。
• 地域分布：按省份统计GMV和订单量，用于可视化地图。

口径统一的意义在于：你前后端展示的所有数字之间要对得上。比如总销售额=日趋势图所有点之和=类目饼图各扇区之和。答辩现场最尴尬的就是被导师发现两个页面数字对不上，一旦对上，你项目可靠性的印象分会直线上升。

3.3 可视化页面规划

我把系统页面规划成三块：

1. 总览看板：顶部四个指标卡（总销售额、总订单数、总销量、客单价），下方左侧类目占比饼图，右侧近30天销售趋势折线图。
2. 商品分析页：Top10热销商品柱状图、价格区间分布图、品牌销量排行。
3. 地域分析页：省份GMV地图、渠道占比（PC/APP/小程序）环形图。

每一块对应一个Flask接口、一组SQL或Spark聚合结果。这样答辩演示时，页面逻辑、后端接口、数据指标三层是一一对应的，讲起来非常清楚。

4. 实操全流程：从环境安装到页面跑通

4.1 环境准备

我用的是虚拟机安装CentOS 7，内存10G，磁盘60G，配置完成后先配好网络和主机名。

# 修改主机名，避免Hadoop启动时hostname解析问题 hostnamectl set-hostname hadoop01 # 关闭防火墙 systemctl stop firewalld systemctl disable firewalld # 配置/etc/hosts，把本机IP映射到hadoop01 echo "192.168.1.100 hadoop01" >> /etc/hosts

安装JDK，解压Hadoop、Spark。这里要先验一下版本：

java -version # 输出应为 1.8.x 或 11.x hdfs version # 输出应为 Hadoop 3.3.x spark-submit --version # 输出应为 Spark 3.3.x

4.2 Hadoop和YARN启动

修改hadoop-env.sh、core-site.xml、hdfs-site.xml、yarn-site.xml。核心就几个参数：

<!-- core-site.xml --> <property> <name>fs.defaultFS</name> <value>hdfs://hadoop01:9000</value> </property>

<!-- hdfs-site.xml --> <property> <name>dfs.replication</name> <value>1</value> </property> <property> <name>dfs.namenode.name.dir</name> <value>/home/hadoop/hadoop_data/namenode</value> </property> <property> <name>dfs.datanode.data.dir</name> <value>/home/hadoop/hadoop_data/datanode</value> </property>

复制为/home/hadoop，不用root跑Hadoop。启动顺序不能乱：

hdfs namenode -format start-dfs.sh start-yarn.sh

启动后检查进程：

jps # 输出里要有 NameNode、DataNode、ResourceManager、NodeManager

4.3 数据生成脚本

我用Python写了generate_data.py，生成商品表、订单表两个CSV，逻辑如下。

import random import datetime import csv categories = [ ("1", "手机数码"), ("2", "家用电器"), ("3", "服饰内衣"), ("4", "美妆护肤"), ("5", "食品饮料"), ("6", "家居家装") ] def gen_products(num=10000): with open("/home/hadoop/dataset/products.csv", "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerow(["product_id", "product_name", "category_id", "category_name", "brand", "price", "shop_id", "status"]) for i in range(1, num + 1): cid, cname = random.choice(categories) price = round(random.uniform(9.9, 9999.0), 2) brand = "模拟品牌" + str(random.randint(1, 100)) writer.writerow([i, f"模拟商品{i}", cid, cname, brand, price, random.randint(1, 500), random.choice([0, 1])])

订单表生成时需要注意两点：付款时间分布在过去365天；pay_price要在商品price基础上乘以折扣。

def gen_orders(product_prices, num=500000): with open("/home/hadoop/dataset/orders.csv", "w", encoding="utf-8", newline="") as f: writer = csv.writer(f) writer.writerow(["order_id", "user_id", "product_id", "quantity", "pay_price", "pay_time", "province", "city", "channel"]) start = datetime.datetime(2023, 1, 1) end = datetime.datetime(2023, 12, 31) provinces = ["浙江", "广东", "江苏", "北京", "上海", "四川", "湖北"] for i in range(1, num + 1): product_id = random.randint(1, 10000) price = product_prices[product_id] discount = random.choice([1, 0.9, 0.8, 0.7, 0.5]) qty = random.randint(1, 5) pay = round(price * discount * qty, 2) ts = start + (end - start) * random.random() province = random.choice(provinces) writer.writerow([ f"O{i:08d}", random.randint(1, 100000), product_id, qty, pay, ts.strftime("%Y-%m-%d %H:%M:%S"), province, province + "市", random.choice(["PC", "APP", "小程序"]) ])

生成后上传到HDFS：

hdfs dfs -mkdir -p /user/hadoop/dataset hdfs dfs -put /home/hadoop/dataset/products.csv /user/hadoop/dataset/ hdfs dfs -put /home/hadoop/dataset/orders.csv /user/hadoop/dataset/

4.4 Spark清洗与指标计算

核心脚本analysis.py用PySpark完成。第一步是创建SparkSession，配置内存参数：

from pyspark.sql import SparkSession from pyspark.sql import functions as F spark = SparkSession.builder \ .appName("taobao_sales_analysis") \ .master("yarn") \ .config("spark.executor.memory", "2g") \ .config("spark.driver.memory", "2g") \ .enableHiveSupport() \ .getOrCreate()

读取订单数据，写死schema能避免CSV类型推断失误：

from pyspark.sql.types import (StructType, StructField, StringType, IntegerType, DoubleType, TimestampType) order_schema = StructType([ StructField("order_id", StringType(), True), StructField("user_id", IntegerType(), True), StructField("product_id", IntegerType(), True), StructField("quantity", IntegerType(), True), StructField("pay_price", DoubleType(), True), StructField("pay_time", TimestampType(), True), StructField("province", StringType(), True), StructField("city", StringType(), True), StructField("channel", StringType(), True) ]) orders_df = spark.read \ .option("header", "true") \ .schema(order_schema) \ .csv("hdfs://hadoop01:9000/user/hadoop/dataset/orders.csv")

清洗逻辑至少要有三步：去除完全重复的订单ID；过滤掉数量小于等于0、金额小于等于0的异常记录；去掉省份为空的数据。

orders_df = orders_df.dropDuplicates(["order_id"]) \ .filter(F.col("quantity") > 0) \ .filter(F.col("pay_price") > 0) \ .filter(F.col("province").isNotNull())

日销售趋势，按天聚合：

daily_df = orders_df.withColumn("date", F.to_date(F.col("pay_time"))) \ .groupBy("date") \ .agg( F.sum("pay_price").alias("gmv"), F.countDistinct("order_id").alias("order_cnt"), F.sum("quantity").alias("sales_vol") ) \ .orderBy("date")

类目排名需要把商品表和订单表Join起来：

products_df = spark.read.option("header", "true") \ .csv("hdfs://hadoop01:9000/user/hadoop/dataset/products.csv") join_df = orders_df.join(products_df, "product_id", "inner") category_df = join_df.groupBy("category_name") \ .agg(F.sum("pay_price").alias("gmv"), F.sum("quantity").alias("sales_vol")) \ .orderBy(F.desc("gmv"))

结果写入MySQL前，先建好结果表。在MySQL里执行：

CREATE DATABASE IF NOT EXISTS taobao_analysis DEFAULT CHARSET utf8mb4; CREATE TABLE IF NOT EXISTS daily_sales ( stat_date DATE PRIMARY KEY, gmv DECIMAL(14,2), order_cnt INT, sales_vol INT ); CREATE TABLE IF NOT EXISTS category_rank ( category_name VARCHAR(100), gmv DECIMAL(14,2), sales_vol INT, ratio DECIMAL(10,4) );

Spark写MySQL：

daily_df.write \ .mode("overwrite") \ .format("jdbc") \ .option("url", "jdbc:mysql://192.168.1.100:3306/taobao_analysis?useSSL=false&serverTimezone=Asia/Shanghai") \ .option("dbtable", "daily_sales") \ .option("user", "root") \ .option("password", "123456") \ .option("driver", "com.mysql.cj.jdbc.Driver") \ .save()

如果这里报错找不到驱动，记得在提交时用--jars mysql-connector-java-8.0.33.jar指定驱动包。

4.5 可视化接口与前端页面

Flask后端只做一件事：从MySQL读结果，转成JSON返回。

from flask import Flask, jsonify import pymysql app = Flask(__name__) def get_conn(): return pymysql.connect( host="192.168.1.100", user="root", password="123456", database="taobao_analysis", charset="utf8mb4" ) @app.route("/api/overview") def overview(): conn = get_conn() cur = conn.cursor() cur.execute("SELECT SUM(gmv), SUM(order_cnt), SUM(sales_vol) FROM daily_sales") total_gmv, total_orders, total_vol = cur.fetchone() cur.execute("SELECT COUNT(*) FROM daily_sales") days = cur.fetchone()[0] avg_price = round(total_gmv / total_orders, 2) if total_orders else 0 conn.close() return jsonify({ "total_gmv": float(total_gmv), "total_orders": int(total_orders), "total_sales_vol": int(total_vol), "avg_price": avg_price, "days": days }) @app.route("/api/trend") def trend(): conn = get_conn() cur = conn.cursor() cur.execute("SELECT stat_date, gmv, order_cnt FROM daily_sales ORDER BY stat_date") rows = cur.fetchall() conn.close() return jsonify([{"date": str(r[0]), "gmv": float(r[1]), "order_cnt": r[2]} for r in rows])

前端页面直接引用ECharts的CDN，用fetch拿数据：

<!DOCTYPE html> <html> <head> <meta charset="utf-8"> <script src="https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js"></script> </head> <body> <div id="trend" style="width:100%;height:400px;"></div> <script> fetch('/api/trend') .then(res => res.json()) .then(data => { var chart = echarts.init(document.getElementById('trend')); chart.setOption({ title: { text: '每日销售额趋势' }, tooltip: { trigger: 'axis' }, xAxis: { type: 'category', data: data.map(d => d.date) }, yAxis: { type: 'value' }, series: [{ name: '销售额', type: 'line', data: data.map(d => d.gmv) }] }); }); </script> </body> </html>

如果你不想用浏览器跨域模式，最简单的办法是直接把HTML放到Flask的templates目录，用render_template渲染，接口和页面同源，没有任何跨域问题。

5. 实操中必踩的五个大坑与解决方案

5.1 NameNode启动不起来

最常见的场景：start-dfs.sh执行完，jps里没有NameNode，或者进程一闪而过。排查路径非常固定：

• hostname和/etc/hosts是否对应，Hadoop通过主机名解析，如果你主机名乱改，NameNode会报UnknownHostException。
• 是否用了root跑Hadoop，如果Hadoop目录属主是普通用户，root启动不一定报错但会把目录权限搞乱。
• 多次namenode -format但是DataNode的current目录还残留旧内容，导致clusterID不匹配，DataNode起不来。遇到这种问题，把hadoop_data整个目录删掉重新格式化，比纠结日志快得多。

5.2 Spark任务总是内存溢出

很多同学在伪分布式上跑大表Join，默认的executor内存只有1G，一聚合就直接java.lang.OutOfMemoryError。建议提交时手动调参：

spark-submit \ --master yarn \ --deploy-mode client \ --driver-memory 2g \ --executor-memory 2g \ --executor-cores 2 \ analysis.py

如果你的脚本里对几百万行做collect()，那再多内存也白搭。尽量只collect聚合后的结果，不要collect原始明细。

5.3 PySpark和Python版本不匹配

pyspark报错的概率极高，提示信息通常很诡异。建议直接用Spark内置的spark-submit提交，并显式设置环境变量：

export PYSPARK_PYTHON=/usr/bin/python3 export PYSPARK_DRIVER_PYTHON=/usr/bin/python3

如果Python是3.9以下，一般没问题；如果错误还是乱，把Python换成3.8或3.9重跑。

5.4 写MySQL时报ClassNotFound

Spark写JDBC，驱动包必须放到Spark的classpath里，光靠pip安装pymysql是不行的，因为那只是Python的MySQL驱动，和Spark JVM侧的JDBC驱动是两回事。

正确做法：把mysql-connector-java.jar放到$SPARK_HOME/jars目录，或者提交参数里写--jars /path/to/mysql-connector-java-8.0.33.jar。另外JDBC URL里必须写useSSL=false&serverTimezone=Asia/Shanghai，否则会报SSL和时区错误。

5.5 图表不渲染

前端最常见的是数据已经有，但图是空白。排查顺序：

• 先打开浏览器的开发者工具，看/api/trend返回的JSON格式是不是数组，字段名和前端取的对不对。
• 看div是否有明确高度，ECharts初始化时容器高度为0会画空白。
• 页面在Flask模板里加载时，fetch的路径要用相对路径，不要写死localhost:5000。

下面是我整理的常见问题速查表，可以保存一份：

6. 答辩准备：怎么讲才能拿高分

6.1 演示顺序

答辩只有十分钟左右的演示时间，顺序一定要设计过。不要从安装环境开始讲，没人爱听“第一步先下载JDK”。

我建议从页面讲起：先打开总览看板，展示四个核心指标和折线图，说“这是全站每天销售额趋势，可以看出来双十一附近有明显波峰”；再切到类目排名图，说“手机数码类目GMV占比最高，接近三成”；接着讲底层链路：“这些结果是由Spark每天对HDFS里的订单明细做聚合，写入MySQL后由后端接口返回给前端。”

讲完后画一条完整的数据流图：Python采集 → HDFS → Spark清洗聚合 → MySQL → Flask → ECharts。只要把这条链讲透，项目主心骨就立住了。

6.2 高频追问与应对思路

导师追问是答辩的重头戏，提前准备好这几个方向：

问：你的数据量有多大？答：模拟生成了约50万条订单和1万条商品，原始数据约300MB，在HDFS里分块存储。如果要扩展到千万级，Spark的并行度是足够的，只需要增加集群节点。

问：为什么用Spark不用Hive？答：Spark提供了更细粒度的ETL能力，特别是多步清洗逻辑用DataFrame API写起来比Hive SQL更灵活；而且Spark的内存迭代计算对分析场景的延迟更低。

问：你的清洗具体清洗了什么？答：去除了订单ID重复的记录，过滤了数量小于等于0或支付金额小于等于0的脏数据，补全/删除了省份为空的数据，并且统一了时间格式和金额精度。

问：出现数据倾斜怎么办？答：如果是Join时某个键的数据量特别大，可以把大键加上随机前缀打散，再二次聚合。我目前的模拟数据里类目分布比较均匀，但没有完全避开这个问题，如果真实环境出现，可以通过加盐方式解决。

问：结果为什么存MySQL，不存HDFS？答：HDFS适合存海量原始数据，但可视化场景需要低延迟随机查询，MySQL对结果数据有索引和事务保障，适合支撑前端接口。这也是经典的“冷热数据分离”思路。

问：如果要实时看数据怎么办？答：可以把订单数据打入Kafka，用Spark Streaming或Flink做实时计算，结果同样落到MySQL或Redis。整个架构可以从离线批处理扩展成离线+实时双链路。

这几个问题答得顺，基本就稳了。

7. 做完之后可以怎么扩展

如果你还有富余时间，我建议在基础版本上做一个小扩展，优先考虑以下方向：

第一，加入用户行为数据，不只是订单，还加上收藏、加购、浏览行为，分析转化率，这会引入漏斗分析模型，项目深度立刻上一个台阶。第二，把清洗后的明细数据落到Hive表，用Hive SQL也能查一遍，这样Hadoop生态的使用痕迹更完整。第三，把可视化图表数量增加到六个以上，加入地域地图、渠道对比、价格带交叉分析，页面饱满度完全不一样。

扩展工作量其实不大，但在答辩时的“研究内容”和“创新点”环节非常加分。尤其是“离线批处理+实时扩展方案”这种话，哪怕是设计上的展望，导师也觉得你考虑到了完整问题。

最后

我做这套项目最大的感受是：大数据毕设真正难的从来不是技术本身，而是“用什么顺序把组件捏合起来”。很多人卡在环境安装，卡在版本兼容，卡在不知道第一步该做设计还是该装软件。如果让我重来一遍，我会坚持先花一下午画清楚表结构、定好指标口径，再把环境搭起来——因为所有代码都是围绕数据和口径打的，前期设计越稳，后面写起来越快，遇到的坑越少。

有个小技巧分享给你：把每个阶段的启动命令和坑点记在同一个笔记里，比如“HDFS格式化后必须删除datanode目录”“Spark提交必须加--jars”。到了写论文的技术实现章节，这些笔记就是现成的素材，比从零回忆省力得多。

希望这份记录能让你少走几个弯路。如果过程中卡在某个报错，多看一眼堆栈日志的开头而不是结尾——多数问题在打印第一行的Caused by里就写明白了。