企业官网建设流程全解析

简介：本资源是一套基于Python+Spark+Hadoop技术栈实现的用户画像驱动型电影推荐系统毕业设计源码案例，面向大数据与人工智能方向的本科高年级学生、毕业设计实践者及初入推荐系统领域的开发者，解决从海量用户行为数据中构建画像、训练模型到生成个性化推荐的全流程工程落地问题。压缩包共802个文件，含60个核心Python脚本（含Spark MLlib建模、Hadoop数据清洗逻辑）、340个前端交互文件（js/css/html等，支撑可视化推荐界面）、151个样式资源及9个SQL数据表结构与示例数据，整体16.2MB，结构完整覆盖后端计算、特征工程、算法集成与前端展示模块。已有82人学习下载，读者可直接获取可运行的全链路代码、分层清晰的目录结构（含data/、src/spark/、web/等标准模块）、用户画像构建逻辑注释、协同过滤与内容推荐双策略实现，以及适配本地伪分布式环境的部署说明，具备强复现性与教学参考价值。

1. 为什么用 Python+Spark+Hadoop 做电影推荐，不是直接上现成的推荐 API？

你手头有一份「Python+Spark+Hadoop大数据基于用户画像电影推荐系统毕业源码案例设计.zip」——别急着解压、别急着 pip install，先问一句：为什么非得拉起一个 Hadoop 集群、再跑 Spark、最后用 Python 写推荐逻辑？
不是因为炫技。而是因为真实场景里，当你的用户量从 1 万涨到 500 万，行为日志从每天 10 万条变成 2 亿条，电影库从 1000 部扩展到 30 万部（含短片、纪录片、UGC 影片），协同过滤矩阵会膨胀到内存根本装不下，单机 Pandas 读 CSV 的速度比人手拷贝还慢，而 Scikit-learn 的NearestNeighbors在 100 万用户 × 50 万物品的稀疏矩阵上跑一次相似度，可能还没出结果，YARN 就给你 kill 掉了。
这个 ZIP 包不是“玩具工程”，它是一套可横向伸缩的工业级推荐链路最小闭环：Hadoop 负责稳住海量原始日志（点击、评分、停留时长、设备类型、地域 IP 段）；Spark 负责在分布式内存里做特征清洗、用户分群、标签打点、隐式反馈加权、ALS 模型训练与实时召回；Python 则作为胶水层，调度任务、封装 REST 接口、做 AB 测试分流、生成用户画像报告 PDF。它解决的不是“怎么推荐一部电影”，而是“怎么让 1000 万活跃用户每人每小时都收到不重复、不冷门、不翻车的个性化片单”。适合正在写毕设但想避开“Flask + SQLite + 随机推荐”雷区的同学，也适合刚转岗数据工程、需要快速理解推荐系统数据底座如何落地的一线工程师。

2. 从零搭起三件套：Hadoop 伪分布式 + Spark Standalone + Python 环境对齐

这不是 Docker Compose 一键拉起的玩具环境。毕业设计/企业 PoC 要求你真正看懂每个组件在链路中承担什么角色、端口怎么撞、配置文件哪一行改错就全链路静默失败。我们按生产调试习惯，用Hadoop 3.3.6（官方编译版） + Spark 3.5.0（预编译包，带 Hadoop 3.3 支持） + Python 3.9.18（conda 管理）组合，全部本地部署，但结构完全对标集群模式。

2.1 Hadoop 伪分布式：不配 ZooKeeper，但必须过hdfs namenode -format这关

伪分布式 ≠ 单机模式。NameNode 和 DataNode 必须是独立 JVM 进程，且core-site.xml和hdfs-site.xml中的fs.defaultFS必须指向hdfs://localhost:9000，而非file:///。常见翻车点：hadoop-env.sh里JAVA_HOME指向了 OpenJDK 17，但 Hadoop 3.3.6 官方只认证 JDK 8/11 —— 启动后jps看不到 NameNode，日志里全是UnsupportedClassVersionError。

# 下载并解压官方二进制包（非源码！） wget https://downloads.apache.org/hadoop/common/hadoop-3.3.6/hadoop-3.3.6.tar.gz tar -xzf hadoop-3.3.6.tar.gz export HADOOP_HOME=$PWD/hadoop-3.3.6 export PATH=$HADOOP_HOME/bin:$PATH # 关键：编辑 $HADOOP_HOME/etc/hadoop/core-site.xml # 替换 <configuration> 内容为： <configuration> <property> <name>fs.defaultFS</name> <value>hdfs://localhost:9000</value> </property> </configuration>

提示：hdfs-site.xml中dfs.replication设为 1（伪分布式只需一份副本），dfs.namenode.name.dir和dfs.datanode.data.dir必须指向绝对路径，且目录需手动mkdir -p并chown给当前用户。格式化命令hdfs namenode -format成功后，$HADOOP_HOME/logs/下应有hadoop-*-namenode-*.log，且末尾出现Storage directory ... has been successfully formatted.

2.2 Spark Standalone：跳过 YARN，用自带 Master/Worker，但必须连通 HDFS

Spark 不是独立运行的。它的spark-defaults.conf里spark.master可设spark://localhost:7077，但spark.sql.warehouse.dir必须指向 HDFS 路径（如hdfs://localhost:9000/user/hive/warehouse），否则CREATE TABLE会报No FileSystem for scheme: hdfs。更隐蔽的坑：Spark 提交任务时默认用本地文件系统读取--jars，若你把mysql-connector-java-8.0.33.jar放在本地，却在代码里写spark.read.jdbc(url, "ratings", ...)，Driver 会连 MySQL，但 Executor 因找不到 jar 包而ClassNotFoundException。

# 下载 Spark 3.5.0（注意选 hadoop3.3 版本） wget https://downloads.apache.org/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz tar -xzf spark-3.5.0-bin-hadoop3.tgz export SPARK_HOME=$PWD/spark-3.5.0-bin-hadoop3 export PATH=$SPARK_HOME/bin:$PATH # 启动 Spark Master（另开终端） $SPARK_HOME/sbin/start-master.sh # 启动 Worker（确保能连上 HDFS） $SPARK_HOME/sbin/start-worker.sh spark://localhost:7077 # 验证：访问 http://localhost:8080，看到 Worker 进程，且 Status 为 Alive # 关键检查：Worker 日志里应有 "Registered signal handlers for [TERM, HUP, INT]" 和 "Successfully registered with master"

2.3 Python 环境：用 conda 锁死版本，避免 PyArrow 与 Spark 冲突

Spark 3.5 默认用 PyArrow 12+ 做向量化 I/O，但pip install pyspark会自动装最新 PyArrow，而 Hadoop 3.3.6 的 libhdfs.so 又要求 PyArrow ≤ 11.0.0。血泪经验：不要用 pip 装 pyspark，用 conda-forge 渠道安装，版本强约束。

# 创建干净环境 conda create -n recsys python=3.9.18 conda activate recsys # 用 conda-forge 安装（关键！） conda install -c conda-forge pyspark=3.5.0 pyarrow=11.0.0 pandas=2.0.3 numpy=1.24.3 scikit-learn=1.3.0 # 验证 PyArrow 兼容性 python -c "import pyarrow as pa; print(pa.__version__); print(pa.hdfs.HadoopFileSystem)" # 应输出 11.0.0 和 <class 'pyarrow._hdfs.HadoopFileSystem'> # 设置 Spark 环境变量（让 Python 找到 Spark JAR） export SPARK_HOME=$PWD/spark-3.5.0-bin-hadoop3 export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-*.zip:$PYTHONPATH

注意：pyspark的 Python API 本质是调用 JVM，所以spark-submit和pyspark shell是等价的。但毕业设计中，你大概率用spark = SparkSession.builder...getOrCreate()启动，此时必须确保SPARK_HOME和PYTHONPATH已导出，否则ImportError: No module named 'py4j'。

3. 用户画像构建：从原始日志到标签宽表，Spark SQL 是主力

用户画像不是“给用户打个‘喜欢科幻’标签”就完事。它是一张动态更新、多粒度、可回溯的宽表：既要支持“最近 7 天点击科幻类 Top3 导演”，也要支持“历史累计评分均值 > 4.2 且观看完成率 > 85% 的高价值用户”。整个流程必须跑在 Spark 上，因为单机处理 10 亿行日志（假设 100 万用户 × 1000 天 × 平均每天 10 条行为）会 OOM。

3.1 原始日志入湖：用 Spark Streaming 模拟实时，但用 Structured Streaming 写 HDFS Parquet

别碰 Kafka 做毕业设计——太重。用 Spark 的socketTextStream或textFileStream模拟实时源，但落盘必须用 Parquet 格式（列存、压缩率高、Schema 自动推断）。关键参数：mergeSchema=true（应对日志字段动态增加），partitionBy("dt")（按天分区，后续查询可裁剪）。

# log_ingest.py：模拟日志接入（实际项目用 Flume/Kafka） from pyspark.sql import SparkSession from pyspark.sql.functions import * from pyspark.sql.types import * spark = SparkSession.builder \ .appName("log-ingest") \ .config("spark.sql.adaptive.enabled", "true") \ .getOrCreate() # 定义 schema（强制指定，避免推断错误） schema = StructType([ StructField("user_id", StringType(), True), StructField("movie_id", StringType(), True), StructField("action", StringType(), True), # click/rating/watch_end StructField("rating", DoubleType(), True), StructField("duration_sec", LongType(), True), StructField("timestamp", TimestampType(), True), StructField("device", StringType(), True), StructField("ip_country", StringType(), True) ]) # 读取实时日志流（本地测试用 textFileStream） logs_df = spark.readStream \ .option("maxFilesPerTrigger", 1) \ .schema(schema) \ .json("hdfs://localhost:9000/input/logs/") # 注意：路径是 HDFS，不是本地 # 写入分区 Parquet 表 query = logs_df.writeStream \ .outputMode("Append") \ .option("checkpointLocation", "hdfs://localhost:9000/checkpoint/log_ingest") \ .partitionBy("dt") \ .format("parquet") \ .start("hdfs://localhost:9000/data/logs_parquet") query.awaitTermination()

逻辑说明：textFileStream会监控 HDFS 目录下新增 JSON 文件，每触发一次读一个文件。partitionBy("dt")要求logs_df中存在dt字段，所以实际代码中需加withColumn("dt", date_format("timestamp", "yyyy-MM-dd"))。checkpointLocation是必须的，Spark 用它记录 offset 和状态，断点续传全靠它。

3.2 标签计算：用 Spark SQL 写离线 ETL，而不是硬编码 UDF

新手常犯错误：用map()写 Python 函数处理每一行，结果性能暴跌。正确做法是全部用 DataFrame API 或 SQL，让 Catalyst 优化器做谓词下推、列裁剪、广播 Join。例如计算“用户最近 30 天活跃度”：

-- user_activity.sql：在 spark-sql CLI 或 SparkSession.sql() 中执行 CREATE OR REPLACE TABLE user_activity AS SELECT user_id, COUNT(*) AS total_actions_30d, COUNT(DISTINCT DATE(timestamp)) AS active_days_30d, AVG(rating) AS avg_rating_30d, -- 计算完播率：watch_end action 数 / click action 数 SAFE_DIVIDE( COUNT(CASE WHEN action = 'watch_end' THEN 1 END), COUNT(CASE WHEN action = 'click' THEN 1 END) ) AS completion_rate_30d, -- 最近一次行为时间 MAX(timestamp) AS last_active_time FROM logs_parquet WHERE dt >= DATE_SUB(CURRENT_DATE(), 30) GROUP BY user_id;

参数说明：SAFE_DIVIDE是 Spark 3.4+ 内置函数，避免除零；DATE_SUB(CURRENT_DATE(), 30)是分区裁剪关键，Spark 会自动只扫描dt在范围内的 Parquet 文件；COUNT(DISTINCT DATE(...))比COUNT(DISTINCT to_date(timestamp))更快，因DATE()是内置函数。这张表每日增量更新，就是用户画像的“事实表”。

3.3 标签宽表组装：用 Broadcast Join 拼接维度表，避免 Shuffle

用户画像最终要导出为一张大宽表（user_profile），包含人口属性（来自注册表）、行为统计（上一步）、内容偏好（下一步）。维度表（如movies表）通常 < 10 万行，用broadcast()强制广播，避免和大表user_activity做 Shuffle Join。

# profile_assemble.py from pyspark.sql.functions import broadcast, when, col, array, sort_array, desc # 读取维度表（小表，广播） movies_df = spark.read.parquet("hdfs://localhost:9000/data/movies/") genres_df = spark.read.parquet("hdfs://localhost:9000/data/genres/") # 广播 Join（显式调用 broadcast()） user_profile_df = user_activity_df \ .join(broadcast(movies_df), "movie_id", "left") \ .join(broadcast(genres_df), "genre_id", "left") \ .groupBy("user_id") \ .agg( # 计算 TOP3 偏好类型（用 collect_list + sort_array） sort_array( array( when(col("genre") == "Sci-Fi", col("total_actions_30d")).otherwise(0), when(col("genre") == "Drama", col("total_actions_30d")).otherwise(0), when(col("genre") == "Comedy", col("total_actions_30d")).otherwise(0) ), asc=False ).alias("preference_score"), # 其他聚合... ) user_profile_df.write.mode("overwrite").parquet("hdfs://localhost:9000/data/user_profile/")

逻辑说明：broadcast()是显式提示，Spark 会将小表序列化后发给每个 Executor，大表分片后直接匹配，省去 Shuffle 阶段。sort_array(array(...))是替代 UDAF 的轻量方案，适合 TOP-K 场景。宽表最终写入 HDFS，供后续 ALS 模型或 Python Flask 接口读取。

4. 推荐模型落地：ALS 协同过滤 + 实时召回，绕开矩阵分解黑匣子

毕业设计最容易被问倒的环节：为什么选 ALS？它和 SVD、LightFM 有什么区别？线上怎么更新？答案不是背概念，而是看数据形态——电影推荐本质是“用户 × 电影”的隐式反馈矩阵（点击/播放/评分），稀疏度 > 99.9%，ALS（Alternating Least Squares）正是为此设计：它不求解完整矩阵，而是交替优化用户因子矩阵 U 和物品因子矩阵 V，使U × V^T ≈ R，且天然支持分布式计算。Spark MLlib 的ALS实现已针对稀疏矩阵优化，比自己用 NumPy 写 SVD 快 10 倍以上。

4.1 数据准备：构造稀疏 Rating RDD，过滤低频用户/物品

ALS 输入必须是(user_id, item_id, rating)三元组。但原始日志中，rating字段只有部分行为有（如action='rating'），其他行为（click,watch_end）需转换为隐式反馈。Spark MLlib 要求user_id和item_id是LongType，且从 0 开始连续编号——这是最大坑点，很多人卡在这步。

# als_preprocess.py from pyspark.ml.feature import StringIndexer from pyspark.sql.functions import when, col, count # 1. 从日志中提取有效交互（隐式反馈：click=1.0, watch_end=2.0, rating=rating） interactions_df = spark.read.parquet("hdfs://localhost:9000/data/logs_parquet/") \ .filter("action IN ('click', 'watch_end', 'rating')") \ .withColumn("rating", when(col("action") == "click", 1.0) .when(col("action") == "watch_end", 2.0) .otherwise(col("rating")) ) \ .select("user_id", "movie_id", "rating") # 2. 过滤：用户至少 5 条行为，电影至少被 10 人交互（去噪） user_cnt = interactions_df.groupBy("user_id").count().filter("count >= 5") movie_cnt = interactions_df.groupBy("movie_id").count().filter("count >= 10") interactions_df = interactions_df \ .join(user_cnt.select("user_id"), "user_id") \ .join(movie_cnt.select("movie_id"), "movie_id") # 3. StringIndexer 编码（关键！必须 fit 后 transform，且 handleInvalid="keep"） user_indexer = StringIndexer(inputCol="user_id", outputCol="user_idx", handleInvalid="keep") movie_indexer = StringIndexer(inputCol="movie_id", outputCol="movie_idx", handleInvalid="keep") indexed_df = user_indexer.fit(interactions_df).transform(interactions_df) indexed_df = movie_indexer.fit(indexed_df).transform(indexed_df) # 4. 转为 ALS 要求的格式（LongType） als_input_df = indexed_df.select( col("user_idx").cast("long").alias("user_id"), col("movie_idx").cast("long").alias("item_id"), col("rating").cast("double").alias("rating") ) als_input_df.write.mode("overwrite").parquet("hdfs://localhost:9000/data/als_input/")

逻辑说明：StringIndexer的handleInvalid="keep"是救命参数——当新用户/电影出现时，它会分配-1，后续 ALS 训练会自动忽略该行。cast("long")必须显式调用，否则 ALS 报java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.Long。

4.2 ALS 训练：调参三要素——rank、maxIter、regParam，以及为什么不用 implicitPrefs

Spark ALS 有两个模式：implicitPrefs=True（隐式反馈）和False（显式评分）。电影推荐中，click和watch_end是隐式信号，但rating是显式信号。混合使用时，必须设implicitPrefs=False，并把隐式行为映射为 1~5 分（如click=2.0,watch_end=4.0,rating=rating），否则模型会把所有行为当作同等强度的“喜欢”，导致推荐泛滥。

# als_train.py from pyspark.ml.recommendation import ALS from pyspark.ml.evaluation import RegressionEvaluator # 读取预处理数据 als_input_df = spark.read.parquet("hdfs://localhost:9000/data/als_input/") # 划分训练/测试集（时间感知划分：用 timestamp 排序，取后 20% 为测试） train_df, test_df = als_input_df.randomSplit([0.8, 0.2], seed=42) # 初始化 ALS（关键参数） als = ALS( userCol="user_id", itemCol="item_id", ratingCol="rating", rank=50, # 隐因子维度，50 是平衡精度与内存的起点 maxIter=10, # 迭代次数，10 足够收敛，再高收益小 regParam=0.01, # L2 正则，防止过拟合，0.01 是经验值 coldStartStrategy="drop", # 新用户/物品不预测，避免 NaN implicitPrefs=False # 显式模式，因我们已映射隐式行为为分数 ) # 训练 model = als.fit(train_df) # 评估 RMSE（用 RegressionEvaluator） evaluator = RegressionEvaluator( metricName="rmse", labelCol="rating", predictionCol="prediction" ) rmse = evaluator.evaluate(model.transform(test_df)) print(f"RMSE on test set: {rmse}") # 合理值：1.2 ~ 1.8（因映射后 rating 范围是 1~5） # 保存模型（HDFS 路径） model.write().overwrite().save("hdfs://localhost:9000/model/als_model_v1/")

参数说明：rank=50是折中选择——rank=10速度快但表达能力弱，rank=200精度略升但内存翻倍；regParam=0.01若设为 0，模型会在训练集上过拟合，测试 RMSE 突然升高；coldStartStrategy="drop"是安全策略，上线后新用户请求会返回空列表，由 Python 层 fallback 到热门榜。

4.3 实时召回：用 model.recommendForAllUsers() 生成全量 TopK，而非在线计算

ALS 模型训练是离线的（每天凌晨跑），但推荐结果要实时响应。常见误区：用户请求时再调model.recommendForUser(user_id, 10)。错！这会触发全量物品打分，延迟秒级。正确做法是离线生成全量用户 Top100 推荐列表，存入 Redis 或 HBase，线上仅查表。

# als_offline_recall.py from pyspark.sql.functions import explode, col # 生成全量用户 Top100 推荐（耗时操作，每日跑一次） user_recs_df = model.recommendForAllUsers(100) # 返回 (user_id, recommendations) 结构 # 展开 recommendations 数组 recs_exploded_df = user_recs_df \ .withColumn("rec", explode("recommendations")) \ .select("user_id", col("rec.item_id").alias("item_id"), col("rec.rating").alias("score")) # 写入 HBase（需配置 hbase-site.xml） recs_exploded_df.write \ .format("org.apache.hadoop.hbase.spark") \ .option("hbase.table", "user_rec") \ .option("hbase.columns.mapping", "user_id STRING :key, item_id STRING cf:item_id, score DOUBLE cf:score") \ .mode("overwrite") \ .save() # 或写入 Redis（用 spark-redis connector） recs_exploded_df.write \ .format("org.apache.spark.sql.redis") \ .option("table", "user_rec") \ .option("key.column", "user_id") \ .mode("append") \ .save()

逻辑说明：recommendForAllUsers(100)是 Spark ALS 最高效召回方式，它利用广播的物品因子矩阵 V，在每个 Executor 上对本分片用户批量计算U[i] × V^T，比逐个用户调用快 100 倍。explode()将数组展开为行，便于写入 KV 存储。线上 Python 服务收到请求，直接redis_client.hgetall(f"user_rec:{user_id}")，毫秒级返回。

5. 避坑指南：那些让你调试三天却只改一行配置的致命细节

毕业答辩前最怕什么？不是算法不收敛，而是pyspark报错信息像天书，或者hdfs dfs -ls显示文件存在，spark.read.parquet()却说Path does not exist。以下是我在 12 个真实项目中踩过的、排名前五的“玄学”坑，每一条都附带现象、根因和一招解决。

5.1 现象：pyspark启动时报java.lang.NoClassDefFoundError: org/apache/hadoop/fs/FileSystem

原因：Spark 的spark-env.sh中HADOOP_CONF_DIR未指向 Hadoop 的etc/hadoop/目录，导致 Spark JVM 找不到core-site.xml和hdfs-site.xml，无法初始化 HDFS FileSystem。
解决：在$SPARK_HOME/conf/spark-env.sh中添加：

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop export YARN_CONF_DIR=$HADOOP_HOME/etc/hadoop

注意：HADOOP_CONF_DIR必须是绝对路径，且目录下必须有core-site.xml。验证方法：spark-submit --conf spark.debug.maxToStringFields=100 --master local[*] your_script.py，看日志是否出现Loaded configuration from ...

5.2 现象：ALS 训练时java.lang.OutOfMemoryError: Java heap space，但spark.executor.memory已设 8g

原因：ALS 的rank参数过大（如设 200），且spark.sql.adaptive.enabled=true时，自适应查询优化器会尝试合并 Stage，导致单个 Task 内存暴增。
解决：关闭 AQE 并调小rank：

spark = SparkSession.builder \ .config("spark.sql.adaptive.enabled", "false") \ # 关键！ .config("spark.executor.memory", "6g") \ .config("spark.driver.memory", "4g") \ .getOrCreate() # ALS rank 改为 30~50

5.3 现象：spark.read.json("hdfs://...")报java.io.IOException: No FileSystem for scheme: hdfs

原因：Spark 的hadoop-client依赖版本与 Hadoop 3.3.6 不兼容。Spark 3.5.0 默认带hadoop-client:3.3.4，但 Hadoop 3.3.6 的libhdfs.soABI 有微小变化。
解决：强制替换 Spark 的 Hadoop JAR：

cd $SPARK_HOME/jars rm hadoop-client-*.jar cp $HADOOP_HOME/share/hadoop/client/hadoop-client-api-3.3.6.jar . cp $HADOOP_HOME/share/hadoop/client/hadoop-client-runtime-3.3.6.jar .

5.4 现象：用户画像宽表中preference_score全是null，但日志里明明有 genre 数据

原因：movies_df和genres_df的 Join key 类型不一致——movies_df.genre_id是StringType，genres_df.genre_id是IntegerType，Spark 静默失败，不报错但不匹配。
解决：统一类型后 Join：

movies_df = movies_df.withColumn("genre_id", col("genre_id").cast("int")) genres_df = genres_df.withColumn("genre_id", col("genre_id").cast("int")) user_profile_df = movies_df.join(genres_df, "genre_id", "left")

5.5 现象：spark-submit提交后，YARN Web UI 显示 Application 状态为ACCEPTED，但永远不变成RUNNING

原因：Hadoop 的yarn-site.xml中yarn.resourcemanager.scheduler.address配置错误，或 ResourceManager 进程未启动。伪分布式下，YARN 并非必须，但若你启用了它，就必须确保start-yarn.sh已执行，且jps能看到ResourceManager和NodeManager。
解决：毕业设计建议直接用 Spark Standalone 模式（spark.master=spark://localhost:7077），彻底绕过 YARN。若必须用 YARN，请检查：

# 确保启动 $HADOOP_HOME/sbin/start-yarn.sh # 检查进程 jps | grep -E "(ResourceManager|NodeManager)" # 检查配置 cat $HADOOP_HOME/etc/hadoop/yarn-site.xml | grep -A2 "scheduler.address" # 应为：yarn.resourcemanager.scheduler.address=localhost:8030

6. 毕业答辩加分项：用 Python Flask 封装推荐接口，并加入 A/B 测试分流逻辑

答辩老师最想看到的不是“我实现了 ALS”，而是“我能让它真正跑起来、被调用、可验证”。所以最后一环，必须把 Spark 训练好的模型，通过 Python Flask 暴露为 REST API，并加入基础的 A/B 测试能力——比如 90% 流量走 ALS 推荐，10% 走热门榜兜底，所有请求日志写入 HDFS，方便后续分析点击率（CTR）。

6.1 构建轻量 API：用 Flask + Redis 缓存，不碰 SparkContext

线上服务绝不能在 Flask 的每个请求里SparkSession.builder.getOrCreate()——那会创建无数 Driver，内存爆炸。正确姿势：Spark 只负责离线生成推荐列表并写入 Redis，Flask 只做查 Redis 和日志记录。

# app.py from flask import Flask, request, jsonify import redis import json import time from datetime import datetime app = Flask(__name__) r = redis.Redis(host='localhost', port=6379, db=0) @app.route('/recommend', methods=['GET']) def recommend(): user_id = request.args.get('user_id') if not user_id: return jsonify({"error": "user_id required"}), 400 # 1. 从 Redis 读取 ALS 推荐（key: user_rec:{user_id}） recs = r.hgetall(f"user_rec:{user_id}") if not recs: # 2. ALS 无结果，fallback 到热门榜（从 HDFS 读取预计算的 top100.csv） # （此处简化：直接返回 mock 热门） hot_list = [{"movie_id": "m1001", "score": 4.8}, {"movie_id": "m2002", "score": 4.7}] strategy = "hot_fallback" else: # 解析 Redis 返回的字节数据 recs_list = [ {"movie_id": k.decode(), "score": float(v.decode())} for k, v in recs.items() ] # 按 score 降序 recs_list.sort(key=lambda x: x["score"], reverse=True) strategy = "als" # 3. A/B 分流：90% ALS，10% 热门（用 user_id 哈希决定） hash_val = sum(ord(c) for c in user_id) % 100 if hash_val < 90: final_recs = recs_list[:10] used_strategy = "als" else: final_recs = [{"movie_id": "m1001", "score": 4.8}, {"movie_id": "m2002", "score": 4.7}] used_strategy = "hot_ab" # 4. 记录日志到 HDFS（异步，避免阻塞） log_entry = { "user_id": user_id, "timestamp": datetime.now().isoformat(), "strategy": used_strategy, "rec_count": len(final_recs), "movie_ids": [r["movie_id"] for r in final_recs] } # 实际项目用 HDFS client 或 Kafka，此处写本地文件模拟 with open("/tmp/recomm_log.jsonl", "a") as f: f.write(json.dumps(log_entry) + "\n") return jsonify({ "user_id": user_id, "recommendations": final_recs, "strategy_used": used_strategy, "server_time": datetime.now().isoformat() }) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=False) # 生产禁用 debug

逻辑说明：r.hgetall()读取的是user_rec:{user_id}这个 Hash 结构，Key 是movie_id，Value 是score（字符串）。hash_val % 100 < 90是简单分流，实际可用一致性哈希。日志写入/tmp/recomm_log.jsonl是为了演示，真实项目应写入 HDFS 的/data/log/recomm/目录，每日按dt分区。

6.2 验证效果：用 curl 测试接口，并用 Spark SQL 分析日志

写完 API，立刻用curl验证，再用 Spark SQL 查日志，形成闭环。这才是工程师思维。

# 测试接口 curl "http://localhost:5000/recommend?user_id=u12345" # 返回示例： # {"user_id":"u12345","recommendations":[{"movie_id":"m5678","score":4.2}],...} # 用 Spark SQL 分析日志（假设日志已上传 HDFS） spark.sql(""" SELECT strategy_used, COUNT(*) as req_count, AVG(rec_count) as avg_rec_count FROM parquet.`hdfs://localhost:9000/data/log/recomm/` WHERE dt = '2024-06-01' GROUP BY strategy_used """).show()

技巧：在app.py中，log_entry加入request.headers.get('X-Real-IP', request.remote_addr)，就能拿到真实 IP，后续可做地域分析。另外，flask默认单线程，生产必须加gunicorn：`gunicorn

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