一、核心架构与适用场景

实时推荐系统是电商、内容平台的核心基建，本文基于Spark MLlib完成离线模型训练（处理海量用户行为数据），结合FastAPI搭建低延迟推理接口，实现“离线模型更新+在线实时推荐”的完整链路。

核心架构拆解

环节	技术选型	核心作用	性能指标
离线数据处理/建模	Spark MLlib	处理千万级用户-物品交互数据，训练协同过滤模型	批处理速度≥10万条/秒
实时推理接口	FastAPI	提供HTTP接口，接收用户ID返回推荐结果	接口响应时间≤50ms
模型存储/加载	Parquet+Redis	离线模型序列化存储，实时特征缓存	模型加载时间≤100ms
部署方式	Docker	环境隔离，一键部署	服务启动时间≤1分钟

适用场景

• 电商商品推荐（基于用户浏览/购买行为）；
• 内容平台文章/视频推荐（基于用户点击/收藏行为）；
• 企业内部资源推荐（基于员工访问记录）。

二、前置准备：环境搭建

1. 核心依赖安装

# 1. 安装Spark（推荐3.3.x版本，兼容MLlib）
# 下载Spark：https://archive.apache.org/dist/spark/spark-3.3.4/spark-3.3.4-bin-hadoop3.tgz
# 配置环境变量（Linux/Mac）
echo "export SPARK_HOME=/usr/local/spark-3.3.4" >> ~/.bashrc
echo "export PATH=\$PATH:\$SPARK_HOME/bin" >> ~/.bashrc
source ~/.bashrc

# 2. 安装Python依赖
pip install pyspark==3.3.4 fastapi==0.104.1 uvicorn==0.24.0.post1 redis==5.0.1 pandas==2.0.3 numpy==1.24.3 scikit-learn==1.3.0
# 验证安装
python -c "import pyspark, fastapi, redis; print('依赖安装成功')"

# 3. 启动Redis（缓存实时特征）
# 安装Redis（Ubuntu示例）
sudo apt update && sudo apt install redis-server -y
# 启动并设置开机自启
sudo systemctl start redis-server
sudo systemctl enable redis-server
# 验证Redis
redis-cli ping  # 输出PONG则正常

2. 数据准备（模拟用户-物品交互数据）

创建user_item_data.csv，包含用户ID、物品ID、交互评分（1-5分，越高表示偏好越强）：

user_id,item_id,rating,timestamp
1001,2001,5,1714521600
1001,2002,4,1714521700
1001,2003,2,1714521800
1002,2001,3,1714521900
1002,2004,5,1714522000
1003,2002,5,1714522100
1003,2005,4,1714522200
1004,2003,3,1714522300
1004,2004,2,1714522400
1005,2001,5,1714522500
1005,2005,5,1714522600

三、Step1：Spark MLlib离线训练推荐模型

1. 训练协同过滤模型（ALS算法）

创建train_model.py，基于交替最小二乘法（ALS）训练用户-物品推荐模型：

from pyspark.sql import SparkSession
from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.sql.types import IntegerType, FloatType, LongType
import os

# 初始化Spark会话
spark = SparkSession.builder \
    .appName("RecommendationModelTraining") \
    .master("local[*]")  # 本地模式，生产环境替换为yarn/spark集群
    .getOrCreate()

# 1. 加载并预处理数据
data_path = "user_item_data.csv"
df = spark.read.csv(data_path, header=True, inferSchema=False)
# 类型转换
df = df.withColumn("user_id", df["user_id"].cast(IntegerType())) \
       .withColumn("item_id", df["item_id"].cast(IntegerType())) \
       .withColumn("rating", df["rating"].cast(FloatType())) \
       .withColumn("timestamp", df["timestamp"].cast(LongType()))

# 2. 划分训练集和测试集
train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)

# 3. 配置ALS模型
als = ALS(
    maxIter=10,  # 迭代次数
    regParam=0.01,  # 正则化参数，防止过拟合
    userCol="user_id",
    itemCol="item_id",
    ratingCol="rating",
    coldStartStrategy="drop",  # 忽略冷启动样本
    seed=42
)

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

# 5. 模型评估
evaluator = RegressionEvaluator(
    metricName="rmse",
    labelCol="rating",
    predictionCol="prediction"
)
predictions = model.transform(test_df)
rmse = evaluator.evaluate(predictions)
print(f"模型测试集RMSE：{rmse:.4f}")  # 越低越好，一般<1表示效果良好

# 6. 保存模型（序列化到本地）
model_path = "./als_recommendation_model"
if os.path.exists(model_path):
    import shutil
    shutil.rmtree(model_path)
model.save(model_path)
print(f"模型已保存至：{model_path}")

# 7. 生成用户推荐结果（测试）
# 为每个用户推荐Top5物品
user_recs = model.recommendForAllUsers(5)
user_recs.show(truncate=False)

# 停止Spark会话
spark.stop()

2. 运行训练脚本

python train_model.py

运行成功后，会输出模型RMSE值，并在本地生成als_recommendation_model目录（包含模型权重和配置）。

四、Step2：FastAPI搭建实时推荐接口

1. 核心接口代码（main.py）

实现“用户推荐”“物品相似推荐”两个核心接口，集成Redis缓存提升响应速度：

from fastapi import FastAPI, HTTPException
from pyspark.sql import SparkSession
from pyspark.ml.recommendation import ALSModel
import redis
import json
import os

# 初始化FastAPI应用
app = FastAPI(title="Real-Time Recommendation API", version="1.0")

# 1. 初始化Redis连接（缓存推荐结果，避免重复计算）
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

# 2. 加载Spark MLlib模型
def load_recommendation_model():
    spark = SparkSession.builder \
        .appName("RecommendationAPI") \
        .master("local[*]") \
        .config("spark.sql.adaptive.enabled", "true")  # 自适应执行优化
        .getOrCreate()
    model_path = "./als_recommendation_model"
    if not os.path.exists(model_path):
        raise FileNotFoundError(f"模型文件不存在：{model_path}")
    model = ALSModel.load(model_path)
    return spark, model

# 全局加载模型（启动时加载，避免每次请求重新加载）
spark, als_model = load_recommendation_model()

# 3. 核心工具函数：生成用户推荐结果
def get_user_recommendations(user_id: int, top_n: int = 5):
    # 先查Redis缓存
    cache_key = f"user_recs:{user_id}"
    cached_recs = r.get(cache_key)
    if cached_recs:
        return json.loads(cached_recs)
    
    # 缓存未命中，调用模型生成推荐
    from pyspark.sql import Row
    user_df = spark.createDataFrame([Row(user_id=user_id)])
    recs = als_model.recommendForUserSubset(user_df, top_n).collect()[0]
    
    # 解析推荐结果
    recommendations = []
    for item in recs.recommendations:
        recommendations.append({
            "item_id": item.item_id,
            "predicted_rating": float(item.rating)  # 转换为Python原生类型
        })
    
    # 缓存结果（过期时间1小时，可根据业务调整）
    r.setex(cache_key, 3600, json.dumps(recommendations))
    return recommendations

# 4. 核心工具函数：生成相似物品推荐
def get_similar_items(item_id: int, top_n: int = 5):
    # ALS模型无直接相似物品推荐，通过物品特征计算余弦相似度
    item_factors = als_model.itemFactors.filter(f"id = {item_id}").collect()
    if not item_factors:
        return []
    
    # 获取目标物品特征
    target_factor = item_factors[0].features
    # 计算所有物品与目标物品的相似度
    all_items = als_model.itemFactors.collect()
    similarities = []
    for item in all_items:
        if item.id == item_id:
            continue
        # 余弦相似度计算
        dot_product = sum(a * b for a, b in zip(target_factor, item.features))
        norm_target = sum(x**2 for x in target_factor)**0.5
        norm_item = sum(x**2 for x in item.features)**0.5
        sim = dot_product / (norm_target * norm_item) if norm_target * norm_item != 0 else 0
        similarities.append((item.id, float(sim)))
    
    # 按相似度排序，取TopN
    similarities.sort(key=lambda x: x[1], reverse=True)
    return [{"item_id": item[0], "similarity": item[1]} for item in similarities[:top_n]]

# 5. API接口定义
@app.get("/api/recommend/user/{user_id}", summary="用户个性化推荐")
def recommend_for_user(user_id: int, top_n: int = 5):
    """
    根据用户ID返回TopN推荐物品
    - user_id: 用户唯一标识
    - top_n: 推荐物品数量，默认5
    """
    try:
        recs = get_user_recommendations(user_id, top_n)
        if not recs:
            raise HTTPException(status_code=404, detail=f"用户{user_id}无推荐结果（冷启动）")
        return {
            "code": 200,
            "msg": "success",
            "data": {
                "user_id": user_id,
                "recommendations": recs
            }
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"推荐失败：{str(e)}")

@app.get("/api/recommend/item/{item_id}", summary="相似物品推荐")
def recommend_similar_items(item_id: int, top_n: int = 5):
    """
    根据物品ID返回相似物品
    - item_id: 物品唯一标识
    - top_n: 相似物品数量，默认5
    """
    try:
        sim_items = get_similar_items(item_id, top_n)
        if not sim_items:
            raise HTTPException(status_code=404, detail=f"物品{item_id}无相似物品")
        return {
            "code": 200,
            "msg": "success",
            "data": {
                "item_id": item_id,
                "similar_items": sim_items
            }
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"相似推荐失败：{str(e)}")

@app.get("/api/health", summary="服务健康检查")
def health_check():
    """验证服务是否正常运行"""
    return {"status": "healthy", "model_loaded": True}

# 6. 启动服务（本地测试）
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host="0.0.0.0",  # 允许外部访问
        port=8000,
        reload=True,  # 开发模式热重载，生产环境关闭
        workers=4  # 多进程，根据CPU核心数调整
    )

2. 启动FastAPI服务

python main.py

服务启动后，可通过以下地址访问：

• 接口文档：http://localhost:8000/docs（自动生成的Swagger文档，可直接测试接口）
• 健康检查：http://localhost:8000/api/health
• 测试用户推荐：http://localhost:8000/api/recommend/user/1001?top_n=5

五、Step3：Docker容器化部署（生产环境）

1. 编写Dockerfile

创建Dockerfile，封装运行环境和代码：

# 基础镜像：Python 3.9（兼容Spark MLlib）
FROM python:3.9-slim

# 设置工作目录
WORKDIR /app

# 安装系统依赖（Spark需要Java）
RUN apt update && apt install -y openjdk-11-jre-headless redis-server && \
    apt clean && rm -rf /var/lib/apt/lists/*

# 下载并配置Spark
ENV SPARK_VERSION=3.3.4
ENV SPARK_HOME=/opt/spark
RUN wget https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop3.tgz -O /tmp/spark.tgz && \
    tar -xzf /tmp/spark.tgz -C /opt && \
    mv /opt/spark-${SPARK_VERSION}-bin-hadoop3 ${SPARK_HOME} && \
    rm /tmp/spark.tgz

# 配置环境变量
ENV PATH=$PATH:${SPARK_HOME}/bin
ENV JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64

# 复制依赖文件并安装
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 复制项目文件
COPY . /app

# 启动脚本
COPY start.sh /app/start.sh
RUN chmod +x /app/start.sh

# 暴露端口
EXPOSE 8000

# 启动服务
CMD ["/app/start.sh"]

2. 编写依赖文件requirements.txt

pyspark==3.3.4
fastapi==0.104.1
uvicorn==0.24.0.post1
redis==5.0.1
pandas==2.0.3
numpy==1.24.3
scikit-learn==1.3.0

3. 编写启动脚本start.sh

#!/bin/bash
# 启动Redis
redis-server --daemonize yes

# 启动FastAPI服务（生产模式）
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

4. 构建并运行Docker镜像

# 构建镜像
docker build -t recommendation-api:v1 .

# 运行容器（后台运行，映射端口）
docker run -d -p 8000:8000 --name rec-service recommendation-api:v1

# 查看容器日志（验证启动状态）
docker logs rec-service

六、性能优化与生产环境适配

1. 接口性能优化

优化点	实现方式
模型加载优化	全局加载模型，避免每次请求重新初始化Spark和加载模型
缓存策略	Redis缓存推荐结果，过期时间根据业务调整（如1小时）
并发优化	FastAPI启用多进程（workers=CPU核心数），Spark配置多线程
数据预处理优化	离线预处理用户/物品特征，避免在线计算

2. 生产环境适配建议

• Spark集群部署：本地模式仅用于测试，生产环境需部署Spark集群（YARN/Standalone），处理TB级数据；
• 模型更新策略：定时（如每天凌晨）重新训练模型，通过符号链接切换模型版本，无需重启API服务；
• 监控告警：集成Prometheus+Grafana监控接口响应时间、QPS、错误率，设置阈值告警；
• 冷启动处理：对新用户/新物品，默认推荐热门物品（离线计算热门榜单），避免返回空结果；
• 接口限流：使用FastAPI-Limiter限制单IP请求频率，防止服务被压垮。

3. 性能测试

使用locust压测接口性能：

# 安装locust
pip install locust
# 创建压测脚本locustfile.py
from locust import HttpUser, task, between

class RecommendationUser(HttpUser):
    wait_time = between(1, 3)  # 每个用户请求间隔1-3秒

    @task
    def user_recommend(self):
        self.client.get("/api/recommend/user/1001?top_n=5")

    @task
    def health_check(self):
        self.client.get("/api/health")

# 启动压测
locust -f locustfile.py --host=http://localhost:8000

访问http://localhost:8089，设置并发用户数和每秒新增用户数，查看接口响应时间、成功率等指标。

七、常见问题与解决方案

问题	原因	解决方案
接口响应时间>50ms	模型加载耗时/无缓存	全局加载模型，增加Redis缓存；优化Spark配置
Docker启动失败（Java缺失）	基础镜像未安装JDK	在Dockerfile中安装openjdk-11-jre-headless
冷启动用户无推荐结果	ALS模型无法处理未交互过的用户	离线计算热门物品，冷启动时返回热门榜单
Redis缓存命中率低	缓存过期时间过短/用户请求分散	调整过期时间（如2小时）；缓存热门用户结果
Spark训练速度慢	本地模式资源不足	部署Spark集群，增加executor内存/核心数