Spark+Flask新能源车数据分析与推荐系统实战：从0到1搭建完整项目

Spark+Flask新能源车数据分析与推荐系统实战：从0到1搭建完整项目

在碳中和政策驱动与新能源汽车产业爆发的双重背景下，车联网数据、用户行为数据、市场交易数据的价值愈发凸显。本文将带大家从零开始，基于Spark完成新能源车数据的分布式分析，再通过Flask搭建轻量级Web服务，结合协同过滤算法实现个性化车辆推荐，最终打造一套完整的新能源车数据分析与推荐系统。全程聚焦实战，所有代码均可直接复用，是大数据工程师、数据分析师的优质实战项目。

• Spark+Flask新能源车数据分析与推荐系统实战：从0到1搭建完整项目
• • 一、项目整体架构与技术栈解析
  • • 1. 核心技术栈
    • 2. 项目流程
  • 二、实战步骤：从数据处理到推荐系统搭建
  • • 1. 环境准备
    • 2. Spark数据处理实战
    • • （1）数据集说明
      • （2）Spark数据清洗与分析
      • （3）运行Spark脚本
    • 3. 基于Spark MLlib实现协同过滤推荐
    • 4. Flask搭建Web服务与API开发
    • 5. 前端可视化展示（简易版）
    • 6. 项目运行与验证
  • 三、项目优化与扩展方向
  • 四、项目价值与应用场景
  • • 总结

一、项目整体架构与技术栈解析

1. 核心技术栈

本次项目的核心技术组合为 Spark + Flask + Python，各组件分工明确：

• Spark：负责海量新能源车数据的清洗、转换、统计分析（如用户充电行为、车型销量分布），利用分布式计算提升处理效率；
• Flask：搭建轻量级Web服务，提供数据查询API和推荐结果展示界面；
• 协同过滤算法：基于用户-车型评分矩阵，实现个性化车辆推荐；
• Echarts：前端可视化，展示数据分析结果（如销量趋势、地区分布）。

2. 项目流程

二、实战步骤：从数据处理到推荐系统搭建

1. 环境准备

确保本地/服务器已安装以下环境：

• Python 3.8+
• Spark 3.3+（配置SPARK_HOME环境变量）
• 依赖库安装：

pip install pyspark flask pandas numpy requests flask-cors

2. Spark数据处理实战

（1）数据集说明

本次使用的数据集包含3个核心文件（可从Kaggle/新能源汽车公开平台获取）：

• user.csv：用户ID、地区、购车预算、充电频率；
• car.csv：车型ID、品牌、续航里程、价格、能耗；
• user_car_score.csv：用户ID、车型ID、评分（1-5分）、浏览时长。

（2）Spark数据清洗与分析

编写spark_data_analysis.py，实现数据清洗和核心指标统计：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, count, sum

# 初始化SparkSession
spark = SparkSession.builder \\
.appName("NewEnergyCarAnalysis") \\
.master("local[*]") # 本地运行，分布式环境可改为yarn
.getOrCreate()

# 1. 加载数据
user_df = spark.read.csv("data/user.csv", header=True, inferSchema=True)
car_df = spark.read.csv("data/car.csv", header=True, inferSchema=True)
score_df = spark.read.csv("data/user_car_score.csv", header=True, inferSchema=True)

# 2. 数据清洗：过滤空值、异常值
# 过滤用户表空值
user_clean_df = user_df.filter(
col("user_id").isNotNull() &
col("region").isNotNull() &
col("budget").between(50000, 500000) # 过滤预算异常值
)

# 过滤车型表空值，续航里程>0
car_clean_df = car_df.filter(
col("car_id").isNotNull() &
col("range_mileage") > 0
)

# 过滤评分表，评分1-5分
score_clean_df = score_df.filter(
col("score").between(1, 5)
)

# 3. 核心指标分析
# 3.1 各地区用户充电频率统计
region_charge_freq = user_clean_df.groupBy("region") \\
.agg(avg("charge_frequency").alias("avg_charge_freq")) \\
.orderBy(col("avg_charge_freq").desc())
print("各地区用户平均充电频率：")
region_charge_freq.show()

# 3.2 不同价格区间车型的平均续航
price_range_car = car_clean_df.withColumn(
"price_range",
# 划分价格区间
when(col("price") <= 100000, "10万以下")
.when((col("price") > 100000) & (col("price") <= 200000), "10-20万")
.when((col("price") > 200000) & (col("price") <= 300000), "20-30万")
.otherwise("30万以上")
).groupBy("price_range") \\
.agg(avg("range_mileage").alias("avg_range")) \\
.orderBy(col("avg_range").desc())
print("不同价格区间车型平均续航：")
price_range_car.show()

# 3.3 构建用户-车型评分矩阵（用于推荐算法）
user_car_matrix = score_clean_df.join(user_clean_df, on="user_id") \\
.join(car_clean_df, on="car_id") \\
.select("user_id", "car_id", "score", "range_mileage", "price")

# 保存处理后的数据为Parquet格式（Spark高效格式）
user_car_matrix.write.mode("overwrite").parquet("data/user_car_matrix.parquet")

# 停止SparkSession
spark.stop()

（3）运行Spark脚本

python spark_data_analysis.py

运行后会输出各地区充电频率、价格区间-续航等核心指标，同时生成user_car_matrix.parquet文件，为后续推荐算法提供数据基础。

3. 基于Spark MLlib实现协同过滤推荐

编写recommendation_algorithm.py，利用Spark MLlib的ALS（交替最小二乘法）实现协同过滤推荐：

from pyspark.sql import SparkSession
from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import RegressionEvaluator

# 初始化SparkSession
spark = SparkSession.builder \\
.appName("CarRecommendation") \\
.master("local[*]") \\
.getOrCreate()

# 加载预处理后的用户-车型矩阵
user_car_matrix = spark.read.parquet("data/user_car_matrix.parquet")

# 划分训练集和测试集（8:2）
(training_data, test_data) = user_car_matrix.randomSplit([0.8, 0.2], seed=42)

# 构建ALS模型
als = ALS(
maxIter=10, # 迭代次数
regParam=0.01, # 正则化参数，防止过拟合
userCol="user_id", # 用户ID列
itemCol="car_id", # 车型ID列
ratingCol="score", # 评分列
coldStartStrategy="drop" # 忽略冷启动（新用户/新车）的NaN值
)

# 训练模型
model = als.fit(training_data)

# 模型评估
predictions = model.transform(test_data)
evaluator = RegressionEvaluator(
metricName="rmse", # 均方根误差
labelCol="score",
predictionCol="prediction"
)
rmse = evaluator.evaluate(predictions)
print(f"模型RMSE（均方根误差）：{rmse:.4f}") # 越小说明模型越准确

# 为指定用户推荐Top5车型
def recommend_cars_for_user(user_id, top_n=5):
# 生成推荐结果
user_recs = model.recommendForAllUsers(top_n)
# 筛选指定用户的推荐结果
user_rec = user_recs.filter(col("user_id") == user_id).collect()
if user_rec:
# 解析推荐结果
rec_list = user_rec[0]["recommendations"]
rec_cars = [(item["car_id"], item["rating"]) for item in rec_list]
return rec_cars
else:
return []

# 测试：为用户ID=1001推荐5款车型
rec_result = recommend_cars_for_user(1001, 5)
print(f"为用户1001推荐的车型ID及预测评分：{rec_result}")

# 保存模型
model.save("model/car_recommendation_model")

spark.stop()

运行脚本后，会输出模型的RMSE（理想值<1），并为用户1001推荐5款车型，同时保存训练好的模型到model/目录。

4. Flask搭建Web服务与API开发

编写flask_app.py，封装数据分析结果和推荐算法，提供REST API：

from flask import Flask, jsonify, request
from flask_cors import CORS
from pyspark.sql import SparkSession
from pyspark.ml.recommendation import ALSModel
import pandas as pd

app = Flask(__name__)
CORS(app) # 解决跨域问题

# 初始化SparkSession（用于加载模型和数据）
spark = SparkSession.builder \\
.appName("FlaskCarRecommendation") \\
.master("local[*]") \\
.getOrCreate()

# 加载训练好的推荐模型
model = ALSModel.load("model/car_recommendation_model")

# 加载车型基础数据（转为字典，方便查询）
car_df = pd.read_csv("data/car.csv")
car_info = car_df.set_index("car_id").T.to_dict()

# 接口1：获取不同价格区间车型的平均续航
@app.route("/api/price_range_range", methods=["GET"])
def get_price_range_range():
# 读取Spark分析结果（也可直接读取本地CSV）
price_range_df = pd.read_csv("data/price_range_car.csv")
# 转为JSON格式
result = price_range_df.to_dict("records")
return jsonify({"code": 200, "data": result})

# 接口2：为指定用户推荐车型
@app.route("/api/recommend", methods=["POST"])
def recommend_cars():
# 获取请求参数
data = request.get_json()
user_id = data.get("user_id")
top_n = data.get("top_n", 5)

if not user_id:
return jsonify({"code": 400, "msg": "用户ID不能为空"})

# 调用推荐函数
def get_recommendations(user_id, top_n):
user_recs = model.recommendForAllUsers(top_n)
user_rec = user_recs.filter(col("user_id") == user_id).collect()
if user_rec:
rec_list = user_rec[0]["recommendations"]
# 补充车型详情
rec_result = []
for item in rec_list:
car_id = item["car_id"]
rec_result.append({
"car_id": car_id,
"pred_score": round(item["rating"], 2),
"brand": car_info[car_id]["brand"],
"range_mileage": car_info[car_id]["range_mileage"],
"price": car_info[car_id]["price"]
})
return rec_result
else:
return []

rec_data = get_recommendations(user_id, top_n)
return jsonify({"code": 200, "data": rec_data})

# 接口3：获取各地区用户充电频率
@app.route("/api/region_charge", methods=["GET"])
def get_region_charge():
region_df = pd.read_csv("data/region_charge_freq.csv")
result = region_df.to_dict("records")
return jsonify({"code": 200, "data": result})

if __name__ == "__main__":
# 启动Flask服务
app.run(host="0.0.0.0", port=5000, debug=True)

5. 前端可视化展示（简易版）

编写index.html，调用Flask API渲染Echarts图表和推荐结果：

<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>新能源车数据分析与推荐系统</title>
<!– 引入Echarts –>
<script src="https://cdn.jsdelivr.net/npm/echarts/dist/echarts.min.js"></script>
<!– 引入jQuery –>
<script src="https://cdn.jsdelivr.net/npm/jquery/dist/jquery.min.js"></script>
<style>
.container {width: 90%; margin: 0 auto;}
.chart {width: 100%; height: 400px; margin: 20px 0;}
.recommend {margin: 20px 0; padding: 20px; border: 1px solid #eee;}
</style>
</head>
<body>
<div class="container">
<h1>新能源车数据分析与推荐系统</h1>

<!– 价格区间-续航图表 –>
<div class="chart" id="priceRangeChart"></div>

<!– 地区充电频率图表 –>
<div class="chart" id="regionChargeChart"></div>

<!– 推荐模块 –>
<div class="recommend">
<h3>个性化车型推荐</h3>
<input type="text" id="userId" placeholder="输入用户ID（如1001）">
<button onclick="getRecommend()">获取推荐</button>
<div id="recResult"></div>
</div>
</div>

<script>
// 初始化价格区间-续航图表
var priceRangeChart = echarts.init(document.getElementById('priceRangeChart'));
$.get("/api/price_range_range", function(res) {
if (res.code === 200) {
var xData = res.data.map(item => item.price_range);
var yData = res.data.map(item => item.avg_range);
priceRangeChart.setOption({
title: {text: '不同价格区间车型平均续航'},
xAxis: {type: 'category', data: xData},
yAxis: {type: 'value', name: '续航（公里）'},
series: [{type: 'bar', data: yData}]
});
}
});

// 初始化地区充电频率图表
var regionChargeChart = echarts.init(document.getElementById('regionChargeChart'));
$.get("/api/region_charge", function(res) {
if (res.code === 200) {
var xData = res.data.map(item => item.region);
var yData = res.data.map(item => item.avg_charge_freq);
regionChargeChart.setOption({
title: {text: '各地区用户平均充电频率'},
xAxis: {type: 'category', data: xData},
yAxis: {type: 'value', name: '充电频率（次/周）'},
series: [{type: 'line', data: yData}]
});
}
});

// 获取推荐结果
function getRecommend() {
var userId = $("#userId").val();
if (!userId) {
alert("请输入用户ID");
return;
}
$.ajax({
url: "/api/recommend",
type: "POST",
contentType: "application/json",
data: JSON.stringify({user_id: parseInt(userId), top_n: 5}),
success: function(res) {
if (res.code === 200) {
var html = "<table border='1'><tr><th>车型ID</th><th>品牌</th><th>续航（公里）</th><th>价格（元）</th><th>预测评分</th></tr>";
res.data.forEach(item => {
html += `<tr>
<td>${item.car_id}</td>
<td>${item.brand}</td>
<td>${item.range_mileage}</td>
<td>${item.price}</td>
<td>${item.pred_score}</td>
</tr>`;
});
html += "</table>";
$("#recResult").html(html);
}
}
});
}
</script>
</body>
</html>

6. 项目运行与验证

python flask_app.py

三、项目优化与扩展方向

四、项目价值与应用场景

总结

✨ 坚持用 清晰的图解 +易懂的硬件架构 + 硬件解析， 让每个知识点都 简单明了 ！ 🚀 个人主页 ：一只大侠的侠 · CSDN

💬 座右铭 ： “所谓成功就是以自己的方式度过一生。”