从毕业设计到实战：手把手教你用Spark MLlib和SpringBoot搭建一个电商推荐系统（附完整源码）

1. 环境准备与架构设计

在开始构建电商推荐系统之前，我们需要先搭建好开发环境。以下是推荐的技术栈组合：

核心组件清单 ：

  • 数据处理引擎 ：Spark 3.2.0（包含MLlib、Spark SQL、Spark Streaming）
  • 后端框架 ：SpringBoot 2.6.3
  • 数据库 ：
    • MongoDB 5.0（主业务数据库）
    • Redis 6.2（实时缓存）
  • 消息队列 ：Kafka 3.1.0
  • 数据采集 ：Flume 1.9.0
  • 开发语言 ：Scala 2.12 + Java 11

系统架构图 ： 

[用户端]
  ↓ HTTP
[SpringBoot服务] → [MongoDB]
  ↓ 日志
[Flume] → [Kafka] → [Spark Streaming]
  ↓              ↓
[Redis]    [Spark MLlib]

提示：建议使用Docker容器化部署各组件，可大幅降低环境配置复杂度。例如通过docker-compose一键启动Kafka+Zookeeper集群。

2. 数据准备与特征工程

2.1 数据模型设计

我们需要三类核心数据表：

MongoDB集合设计 ： 

// 商品表
{
  "productId": "B00006HAXW",
  "name": "无线蓝牙耳机",
  "categories": ["电子产品","音频设备"],
  "tags": ["降噪","运动","Hi-Res"],
  "price": 299.00
}

// 用户评分表
{
  "userId": "U1001",
  "productId": "B00006HAXW",
  "rating": 4.5,
  "timestamp": 1659321600
}

// 用户行为表
{
  "userId": "U1001",
  "productId": "B00006HAXW",
  "behavior": "click", // 支持click/cart/purchase等
  "duration": 30,      // 停留时长(秒)
  "timestamp": 1659321600
}

2.2 特征提取策略

对于协同过滤算法，我们需要构建用户-商品交互矩阵： 

// Scala代码示例：构建评分矩阵
val ratings = spark.read.mongo("mongodb://localhost:27017/ecommerce.ratings")
  .select($"userId", $"productId", $"rating")
  .rdd
  .map { row =>
    (row.getAs[String]("userId").hashCode.toInt, 
     row.getAs[String]("productId").hashCode.toInt,
     row.getAs[Double]("rating"))
  }

特征权重分配表 ：

行为类型 权重 说明
点击 0.3 表示初步兴趣
加购 0.6 强购买意向信号
购买 1.0 最明确的偏好表达
收藏 0.8 长期兴趣指标

3. 核心算法实现

3.1 离线推荐模块

ALS协同过滤实现 
// 配置ALS参数
val als = new ALS()
  .setRank(50)          // 隐特征维度
  .setMaxIter(20)       // 迭代次数
  .setRegParam(0.01)    // 正则化参数
  .setUserCol("userId")
  .setItemCol("productId")
  .setRatingCol("rating")

// 训练模型
val model = als.fit(training)

// 生成推荐
val userRecs = model.recommendForAllUsers(10) // 每个用户推荐10个商品

参数调优技巧 ：

  1. 使用交叉验证确定最佳rank值： 
val paramGrid = new ParamGridBuilder()
  .addGrid(als.rank, Array(10, 50, 100))
  .build()

val evaluator = new RegressionEvaluator()
  .setMetricName("rmse")

val cv = new CrossValidator()
  .setEstimator(als)
  .setEvaluator(evaluator)
  .setEstimatorParamMaps(paramGrid)
  .setNumFolds(3)

val cvModel = cv.fit(training)
冷启动解决方案

对于新用户/新商品，采用混合推荐策略：

  1. 基于内容的推荐（商品标签匹配）
  2. 热门商品兜底
  3. 知识图谱关联推荐 
# Python示例：内容相似度计算
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

product_descriptions = ["蓝牙 无线 耳机 降噪", ...]
vectorizer = TfidfVectorizer()
tfidf_matrix = vectorizer.fit_transform(product_descriptions)
cosine_sim = cosine_similarity(tfidf_matrix, tfidf_matrix)

3.2 实时推荐模块

实时处理流水线 ： 

用户行为 → Flume采集 → Kafka → Spark Streaming → 实时推荐引擎 → Redis

// Spark Streaming处理逻辑
val kafkaStream = KafkaUtils.createDirectStream[...](
  ssc, 
  LocationStrategies.PreferConsistent,
  ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
)

kafkaStream.map { record =>
  // 解析用户行为
  val (userId, productId, behavior) = parseLog(record.value())
  
  // 从Redis获取用户最近行为
  val recentActions = redisClient.lrange(s"recent:$userId", 0, 5)
  
  // 计算实时推荐得分
  calculateRealtimeScore(userId, productId, behavior, recentActions)
}

实时评分算法 ： 

实时得分 = 基础分 × 时间衰减系数 × 行为权重
其中：
- 基础分：ALS预测分或内容相似度分
- 时间衰减：1 / (1 + log(时间差))
- 行为权重：点击0.3，购买1.0等

4. 系统集成与API设计

4.1 SpringBoot接口设计 

@RestController
@RequestMapping("/api/recommend")
public class RecommendController {
    
    @GetMapping("/forUser")
    public List<Product> getUserRecommendations(
        @RequestParam String userId,
        @RequestParam(defaultValue = "10") int size) {
        
        // 从Redis获取实时推荐
        List<String> realtimeRecs = redisTemplate.opsForList()
            .range("rec:" + userId, 0, size - 1);
            
        // 若实时推荐不足，补充离线推荐
        if(realtimeRecs.size() < size) {
            List<String> offlineRecs = mongoTemplate.find(
                Query.query(Criteria.where("userId").is(userId))
                    .limit(size - realtimeRecs.size()),
                UserRec.class);
            realtimeRecs.addAll(offlineRecs);
        }
        
        return productService.getProducts(realtimeRecs);
    }
}

4.2 性能优化方案

缓存策略 ：

  1. 使用Redis缓存热门推荐结果
  2. 采用增量更新机制：
    • 离线推荐每天全量更新
    • 实时推荐每分钟增量更新
  3. 多级缓存设计： 
    @Cacheable(value = "userRecs", 
           key = "#userId",
           unless = "#result == null || #result.size() < 5")
public List<Recommendation> getRecommendations(String userId) {
    // ...
}

Spark调优参数 ： 

spark.executor.memory=8G
spark.driver.memory=4G 
spark.serializer=org.apache.spark.serializer.KryoSerializer
spark.sql.shuffle.partitions=200

5. 部署与监控

5.1 容器化部署

docker-compose.yml 示例片段： 

services:
  spark-master:
    image: bitnami/spark:3.2.0
    ports:
      - "8080:8080"
    environment:
      - SPARK_MODE=master

  spark-worker:
    image: bitnami/spark:3.2.0
    depends_on:
      - spark-master
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark-master:7077

5.2 监控指标

需要监控的关键指标：

  1. 推荐质量 ：

    • 点击通过率(CTR)
    • 转化率(CVR)
    • 推荐多样性

  2. 系统性能 ： 

    # Spark监控
curl http://spark-master:4040/api/v1/applications

# Redis监控
redis-cli info stats | grep instantaneous_ops_per_sec

  3. 业务看板 ： 

    -- MongoDB聚合查询示例
db.ratings.aggregate([
  { $match: { timestamp: { $gt: lastHour } } },
  { $group: { 
      _id: "$productId", 
      avgRating: { $avg: "$rating" },
      count: { $sum: 1 }
  }}
])

6. 完整源码结构

项目目录结构： 

ecommerce-recommender/
├── spark-jobs/          # Spark作业
│   ├── offline/         # 离线推荐
│   └── streaming/       # 实时处理
├── recommender-api/     # SpringBoot服务
│   ├── src/
│   └── Dockerfile
├── docker-compose.yml   # 容器编排
└── datasets/            # 示例数据
    ├── products.json
    └── ratings.csv

关键代码文件：

  1. ALSRecommender.scala - 离线推荐主逻辑
  2. RealtimeStreaming.scala - 实时处理流水线
  3. RecommendService.java - SpringBoot服务层
  4. RecsysMonitor.java - 监控端点实现

提示：完整源码已托管在GitHub，包含详细配置说明和测试数据集，可通过仓库README获取部署指南。

7. 常见问题排查

问题1：Spark作业执行缓慢

  • 检查数据倾斜： 
    ratings.groupBy("userId").count().orderBy($"count".desc).show(10)
  • 解决方案：
    • 增加 spark.sql.shuffle.partitions
    • 对倾斜键单独处理

问题2：实时推荐延迟高

  • 优化检查点设置： 
    ssc.checkpoint("hdfs://checkpoints")
  • 调整Kafka消费参数： 
    spark.streaming.kafka.maxRatePerPartition=1000

问题3：推荐结果重复率高

  • 引入多样性策略： 
    val diversifiedRecs = recs
  .groupBy(_.category)
  .flatMap { case (_, items) => 
    items.take(2) // 每类最多取2个
  }
  .take(10)

在实际部署过程中，建议先使用小规模数据进行端到端测试，验证各组件连通性后再逐步放大数据规模。对于生产环境，需要考虑添加灰度发布和A/B测试能力来验证推荐效果。

# 机器学习
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

基于STM32的全屋环境智能监测与预警终端设计

avatar 亚马逊云科技技术品牌专区

Keras：面向人类的深度学习库

Keras3支持多后端（TensorFlow、PyTorch、JAX），通过设置环境变量切换，兼容指定版本。pip安装后可用Sequential API快速构建CNN模型，如MNIST分类示例，实

avatar 亚马逊云科技技术品牌专区

AI行业日报｜2026-09-25：5个热点事件

本期行业动态聚焦于端侧算力重构、自主智能体工作流、开发工具经济模型、网络隔离架构设计与公共部门合规审查。以下结合一线技术场景进行结构化梳理，供 AI 研究员与软件工程团队参考。 PrismML 正将其轻量级语言模型适配至搭载高通处理器的智能眼镜终端 1 。该平台的核心战略指向开源权重模型的本地化运行，旨在最大化挖掘穿戴设备现有算力的利用效率 1 。对于嵌入式与边缘计算开发者而言，此类微型模型向低功

avatar 亚马逊云科技技术品牌专区