做这套基于 Spark Hadoop Python 的商品房推荐系统时我心里其实就一个想法把用户从海量房源里捞出来不用再一页一页翻。房源数据到了千万级用户行为日志每天上百万条单机 Python 脚本根本扛不住必须上分布式。这套系统就是用 Hadoop 做底层存储Spark 做内存计算PySpark 做特征工程和模型训练最后用 Python 包装成可用的排序结果直接对应前端展示。整套链路下来核心解决的问题有三个数据量大了怎么算得动用户行为稀疏怎么办以及推荐结果怎么真正贴合“房子”这种低频高客单商品的特殊性。不管你是做大数据开发的还是刚转推荐系统方向这篇文章值得你花十分钟看完。我不讲花哨的算法理论只讲这套系统从数据清洗到模型落地的每一步怎么走、中间踩了哪些坑、参数怎么调才有效。1. 整体架构与选型思路1.1 为什么偏偏是 Spark Hadoop Python很多团队做推荐系统一上来就纠结技术栈。如果你处理的房源量只有几十万线上用户几万人那坦白说 MySQL 加 Redis 加一个 Python 脚本就够用了根本不用上 Spark。但我面对的情况不一样房源数据来自多个渠道加上历史成交和用户点击行为数据量级在几千万行特征上百个单机跑一次特征聚合得几个小时而且每天都要更新。这时候你需要的不是一台机器而是一个能横向扩展的计算集群。Hadoop HDFS 负责存原始数据。日志文件、房源快照、用户行为明细这些数据一旦落盘基本不会修改HDFS 的流式读写在批处理场景下优势很明显。Spark 负责算。为什么不用 Hadoop 原生的 MapReduce因为推荐模型训练是个迭代过程ALS 矩阵分解每一轮迭代都要反复读取中间结果MapReduce 每轮都落磁盘速度慢得让人怀疑人生。Spark 基于内存计算配合 DAG 调度器同一份数据可以在内存里被多个 Stage 复用迭代效率高一个量级。Python 在这个架构里充当“胶水层”加“算法层”。PySpark 让你用 Python 语法操作 Spark DataFrame写特征工程的时候比写 Scala 舒服得多。模型部分用 Spark MLlib 的 ALS做完训练后把结果导出来再用 Python 做规则融合和排序服务。相当于分布式的重活累活交给 Spark灵活的规则逻辑交给 Python各取所长。1.2 系统分层与完整数据链路我习惯把系统拆成四层来看每层职责单一出了问题也好排查。数据接入层用户行为埋点日志、房源信息表、成交记录通过定时任务同步到 HDFS按天分目录存储。数据加工层用 PySpark 做清洗、去重、特征工程生成用户特征宽表和房源特征宽表注册成 Hive 表供后续查询。模型层召回模型协同过滤 基于内容 热门兜底和排序策略输出每个用户的候选房源列表和得分。服务层把排序结果写入 Redis 或者 MySQL线上接口直接读取给前端和 App 提供数据。这条链路设计成离线为主、在线为辅。离线批量计算每天跑一次把用户当天需要看到的推荐结果预生成好。在线部分只做轻量交互比如用户进入页面时读取结果、点击后上报行为日志。对于房产这种决策周期长、用户不会频繁刷的品类离线计算完全够用没必要上实时 Flink 那套复杂架构。注意这套架构不是万能的。如果你的产品做的是高频快消品类用户几分钟内就会产生大量新行为离线 T1 的推荐就滞后了需要引入实时计算组件。房产低频高客单离线足够。1.3 单机方案和分布式方案的取舍对比我画了一张对比表方便你对照自己的场景做判断。对比维度单机 Python MySQLSpark Hadoop Python数据量支撑百万级以内亿级特征计算效率跑一次数小时分布式并行分钟级模型训练支持需自己实现或调库MLlib 内置 ALS分布式训练扩容能力加内存、换 CPU有限加节点水平扩展运维复杂度低中高需要集群适用场景初创期、数据量小数据量大、特征复杂、团队有维护能力建议别盲目上分布式。如果数据量一天几十万条单机方案跑得又快又省钱没必要给自己找运维负担。但是一旦数据量过了千万或者你每天要训练多轮模型分布式带来的时间节省是肉眼可见的。我实测下来同一份量级的训练任务单机 Python 要跑三个多小时的Spark 集群上十分多钟出结果。2. 数据工程——推荐系统真正的重头戏2.1 数据采集明确要什么行为、存什么字段很多推荐项目挂掉不是模型不行是数据没攒好。做商品房推荐数据至少要有三块房源属性数据、用户画像数据、用户行为数据。房源属性包括楼盘名称、所在城市、区域板块、总价、单价、面积、户型、朝向、装修状态、是否有电梯、楼龄、周边配套标签地铁、学校、商超。这些都是结构化数据直接从业务库同步到 HDFS 就行。关键的是行为数据你得决定采集哪些行为。我用的是四类浏览、收藏、电话咨询、预约看房。这四类行为的程度是递进的咨询和看房说明意向极强浏览只是随手看看。行为日志至少要包含这些字段用户 ID、房源 ID、行为类型、行为时间戳、停留时长、来源渠道搜索页、推荐位、首页。来源渠道这个字段容易被忽略但它对推荐效果影响很大。我后面在排序阶段会用渠道信息做加权因为用户从推荐位点进来和从搜索页点进来意图强度完全不同。2.2 数据清洗把脏数据挡在特征工程之前数据清洗是最不性感但最重要的一步。HDFS 上的原始日志脏得超出想象用户 ID 为空、房源 ID 不存在、时间戳格式五花八门、同一事件被上报了三次。我的清洗规则比较直接去重按用户 ID、房源 ID、行为类型、时间戳做精确去重防止网络抖动导致重复上报。过滤无效数据用户 ID 或房源 ID 为空的直接丢弃房源在房源表中不存在的丢弃避免特征 join 时产生大量空值。时间规范化统一转成时间戳格式后面做时间窗口划分要用。异常值剔除价格字段 0 或者极端的大于市场价 100 倍直接剔除实收面积小于 5 平米的剔除。这些异常值会严重干扰协同过滤的相似度计算。这一步用 PySpark 写很简单关键是别跳过。我见过不少团队图省事数据不洗直接送模型结果离线评估指标还挺高上线以后效果一塌糊涂就是因为脏数据产生了虚假相关性。2.3 特征工程用 PySpark 高效生成宽表特征决定上限算法逼近上限。这句话在推荐系统里是铁律。商品房的用户行为稀疏用户不会天天看房所以要特别用心做特征。用户特征方面我做的是统计类 偏好类用户浏览房源的总数、平均看房预算、偏好面积段、偏好区域 TOP3用户近 7 天、30 天活跃度浏览天数、浏览次数用户最近一次行为距今多少天超过 30 天说明已经流失房源特征方面除了自身的属性特征还要算“相对特征”房源总价相对同区域均价的比例高于 1.2 倍算偏高房源单价在同类户型中的分位排名该房源在近 7 天被浏览、收藏、咨询的次数这一类特征最关键的是一个“动态性”。房源属性是静态的但热度是动态变化的用户的行为偏好也是变化的。所以特征表需要每天更新。PySpark 代码示例from pyspark.sql import SparkSession from pyspark.sql import functions as F spark SparkSession.builder \ .appName(house_rec_feature) \ .enableHiveSupport() \ .getOrCreate() # 读取行为日志 behavior spark.table(ods_user_behavior_log) # 用户偏好特征 user_pref behavior.groupBy(user_id).agg( F.count(house_id).alias(total_view_cnt), F.avg(total_price).alias(avg_price_pref), F.collect_set(district).alias(district_list) ) # 房源热度特征 house_hot behavior.filter(F.col(event_type) view) \ .groupBy(house_id).agg( F.count(*).alias(view_cnt_7d), F.count(F.when(F.col(event_type) collect, 1)).alias(collect_cnt_7d) ) # 特征关联 user_feature user_pref.join(house_hot, onuser_id, howleft) user_feature.write.mode(overwrite).saveAsTable(dws_user_feature)实操提示groupBy 和 join 是最常见的两个算子一定要关注数据倾斜。如果某个热门房源的浏览计数特别大会导致一个 task 处理的数据量远大于其他 task。后面第 5 章我会专门聊解决思路。2.4 为什么特征必须离线批量算在线实时计算特征听起来很酷但对于商品房推荐来说性价比太低。实时算一个用户的偏好需要把用户历史所有行为拉出来聚合计算一次。一个用户历史行为可能有几千条几千个用户同时请求系统压力非常大。而且房产推荐根本不需要秒级更新——用户今天看房明天再看后天你的推荐变了这完全是合理节奏。我的做法是每天凌晨 2 点 Spark 跑批生成当天的特征宽表覆盖写入 Hive。线上推荐时直接查宽表拿用户特征、房源特征和模型输出的候选集几分钟内完成。这样既准又省。3. 推荐算法实现与离线评估3.1 召回层先用 ALS 协同过滤找到可能感兴趣的房源协同过滤是大路货但也是召回阶段性价比最高的方法。我用的 Spark MLlib 里的 ALS交替最小二乘法。ALS 本身处理的是“用户对物品评分”但我们的场景里没有显式评分只有隐式反馈信息。浏览算 1 分收藏算 2 分咨询算 5 分预约看房算 8 分不存在的交互算 0 分。这样把行为映射成一个“隐式评分矩阵”。这里有个关键参数alpha。在隐式反馈模型里alpha控制置信度。我调参时发现如果alpha设太小比如 0.01模型会把所有用户都往热门房源上拉因为热门房源被点击得多、置信度高。设太大比如 10又容易过拟合到个别行为上。我最终用的是alpha1.0配合rank50、regParam0.01、maxIter15AUC 和 NDCG 都稳定在较好水平。from pyspark.ml.recommendation import ALS als ALS( userColuser_id, itemColhouse_id, ratingColimplicit_score, rank50, regParam0.01, alpha1.0, maxIter15, implicitPrefsTrue, coldStartStrategydrop ) model als.fit(train_data) user_recs model.recommendForAllUsers(50)选型心得ALS 之所以适合这个场景是因为矩阵分解能捕捉用户和房源的潜在因子比如“预算 200 万左右、偏好地铁沿线、两居室”这样的隐语义。但列出的潜在因子没有显式含义你不用纠结“因子 3 代表是不是学区房”这只是数学上的抽象表示。3.2 内容召回按用户最近的偏好找相似房源协同过滤有个痛点新上架的房源没有任何行为数据矩阵分解根本没法给它算出向量表示。这时候就要靠基于内容的召回兜底。我维护了一套房源属性向量包括区域、总价档位、面积档位、户型、朝向、装修、地铁距离等类别特征做 One-Hot 编码。用户最近的浏览记录里选出最近 5 个房源取它们的属性向量均值作为用户“近期偏好向量”然后计算这个向量和全量房源向量的余弦相似度取 TOP N 作为内容召回候选。这里有一个很实用的细节价格相似度跟现实中的“预算”概念不太一样。用户看了一次 600 万的房子不代表他买得起 600 万更不代表他想看 600 万的。我用的是“同价位段”的逻辑把价格离散成 50 万一个档位用户看过的房子所在档位的邻档也纳入相似范畴这样既不会死板也不会离谱。3.3 排序层多路召回如何融合排序召回回来的候选集有三个来源ALS 协同过滤、内容相似、热门兜底。每一路的得分量纲不同不能直接相加。我用的是简单有效的加权融合final_score 0.4 * als_score_norm 0.3 * content_score_norm 0.2 * popularity_score_norm 0.1 * distance_bonus这里的als_score_norm是把 ALS 预测分做 min-max 归一化。popularity_score是房源近 7 天浏览量的对数变换再归一化。distance_bonus是用户所在商圈与房源所在商圈的匹配加分如果在同一板块 1同城不同板块 0.5不同城市直接不参与排序。调权重的过程没有捷径我一开始设的是 0.4 / 0.3 / 0.2 / 0.1后来发现热门权重占比太大导致推荐结果偏热门、个性化不足。调整成 0.5 / 0.3 / 0.1 / 0.1 之后覆盖率提升明显。建议你拿一星期的 AB 测试数据来决定权重组合不要拍脑袋。3.4 离线评估别被看着漂亮的指标骗了离线评估我用三个指标Precision10、Recall10、NDCG10。Precision 衡量推荐列表有多少被用户真实点过Recall 衡量用户点过的房源有多少被推荐出来了NDCG 衡量排序质量——好的结果是不是排前面了。划分数据有个大坑不能用随机划分。用户 1 到 10 号的行为在训练集里15 号的行为在测试集里这没问题。但如果同一个用户的同一时间段行为被随机分到训练和测试里就发生了数据穿越模型相当于“看到过答案再考试”离线指标虚高。正确的做法是按时间切分前 30 天的行为做训练最近 14 天的行为做测试。离线指标高不代表线上效果好这是老生常谈。离线指标只能帮你筛掉明显不行的模型真正的试金石还是线上小流量 AB。我见过离线 NDCG 提升 0.05 的模型上线后点击率反而降了因为离线环境无法模拟用户在真实场景下的行为链条。但至少离线评估能帮你拦住很多低级错误该做的环节省不了。4. 推荐结果落地的工程实践4.1 用 spark-submit 提交离线任务整个推荐流程最终是一个 PySpark 脚本提交到 YARN 集群上跑。我的提交命令供参考spark-submit \ --master yarn \ --deploy-mode cluster \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 10 \ --driver-memory 2g \ --conf spark.sql.shuffle.partitions200 \ run_recommend.py这些参数不是随便拍的。executor-memory 4g是为了平衡内存占用和 GC 开销设太大容易出现长时间的 Full GC。executor-cores 2是因为模型训练阶段每个任务并不是纯计算还要等待磁盘和网络 IO核数太多容易跟内存争抢。spark.sql.shuffle.partitions200是为了避免默认 200 个分区导致单个分区数据量过大后面讲数据倾斜时还会提到。4.2 从训练到写库全流程代码骨架推荐脚本主要步骤是从 Hive 读取特征宽表、划分训练测试集、训练 ALS 模型、生成用户推荐候选、内容召回和热门候选融合、排序、把结果写入 MySQL 和 Redis。from pyspark.sql import SparkSession import pymysql spark SparkSession.builder.appName(house_recommend).enableHiveSupport().getOrCreate() # 读取特征表 user_feature spark.table(dws_user_feature) house_feature spark.table(dws_house_feature) # 排序结果写 MySQL用 DataFrame 转成 Python list批量写入 recommend_results sorted_result.collect() conn pymysql.connect(hostyour_host, useryour_user, passwordyour_pwd, databaserec_db) cursor conn.cursor() for row in recommend_results: cursor.execute( INSERT INTO t_user_recommend (user_id, house_id, score) VALUES (%s, %s, %s), (row.user_id, row.house_id, row.score) ) conn.commit()提示从 Spark 往 MySQL 大批量写数据时别一条条写性能差得离谱。积攒一批再批量写或者直接生成 CSV 用LOAD DATA导入。线上接口读的是 RedisMySQL 只是备份和离线分析用。4.3 增量更新策略与冷启动处理房源数据每天都会有变化新盘上架、房子下架、价格微调。用户行为也在持续累积。我的策略是每天全量训练一次模型。你可能会觉得全量训练浪费算力但以房源推荐的数据量级Spark 集群跑一次全量 ALS 也就十几分钟完全在可接受范围内。增量训练反而要维护版本状态复杂度高、收益不明显除非数据量真的到了亿级日增。冷启动分两种。第一是新用户只有注册信息和极少行为我的策略是热门房源兜底 同商圈竞价排名。第二是新上架的房源没有行为数据进不了协同过滤候选就靠内容召回的相似度计算推荐给“潜在偏好相似”的用户——具体做法是给新上架房源打上“新品加权”在内容召回阶段给它额外乘一个新品系数这样用户有机会看到刚上架的房源不至于沉底。4.4 效果监控怎么落地推荐系统上线只是开始监控才是持久战。每天看三个核心指标曝光覆盖率当日推荐结果中有多少房源被用户看到了、点击率、平均点击位置。点击率下降说明推荐质量可能有问题点击位置集中在第一名说明排序不够合理。另外要监控一个容易被忽视的指标用户推荐结果的新鲜度。如果连续一周推荐列表基本没变用户会产生“系统根本不懂我”的疲劳感。我在排序公式里加入了时间衰减权重用户 7 天前看过的房源排序时会适当降低权重保证每天打开 App 都有新东西。5. 常见问题与排查技巧实录5.1 数据倾斜怎么破热点房源压垮单个节点这是我在整个项目里遇到最多的坑。某个核心商圈的头部楼盘浏览量是普通房源的几百倍。在 groupBy 和 join 时这些热点 key 全部落在同一个分区上导致一个 Executor 忙死其他 Executor 闲得看戏任务卡住大半天。常用的解法有三个加盐、广播小表、提高分区数。加盐是把原始的 key 加上随机前缀先分散到不同分区做局部聚合再去掉前缀做全局聚合这个方法对 groupBy 特别有效。广播小表适合 join 场景如果一个小表只有几十 MB用 broadcast join 避免 shuffle。另外把spark.sql.shuffle.partitions从 200 提到 400 也能缓解但治标不治本。5.2 Executor OOM内存溢出的排查路径训练过程中最容易出现的是 Executor 内存溢出报错java.lang.OutOfMemoryError。原因各有不同我梳理了一个排查顺序先看是不是executor-memory设太小4g 是起步值。再看是不是有大对象比如用户特征一次性加载太多考虑减少每个分区的一次性处理量。排查shuffle阶段内存压力把spark.memory.offHeap.enabled开启配合spark.memory.offHeap.size2g做堆外内存缓冲。考虑调整StorageLevel把中间 RDD 存到磁盘而不是内存。最笨但有效的方法是缩小输入数据量先跑一个小分区的测试集验证逻辑无误后再上全量数据。5.3 模型训练时间过长参数维度大、收敛慢怎么处理ALS 在大规模稀疏矩阵上训练rank 太大或者 maxIter 太多训练时间会明显拉长。我的实测经验是rank50对房源推荐已经足够提高到 100 的收益几乎可以忽略训练时间却翻了一倍。另外如果你发现训练不收敛可以先检查输入评分矩阵是不是过于稀疏——超过 99.9% 为 0 的矩阵最有效的做法是考虑降低用户维度比如只保留有过 3 次以上行为的用户。训练时间还跟数据倾斜有关前面提到的热点房源会拖慢每次迭代的速度。把倾斜问题解决后训练时间能缩短 40% 以上。5.4 数据穿越离线评估指标的隐形杀手我前面提到过训练测试集必须按时间切分。但还有一种更隐蔽的数据穿越测试集中的用户行为被用于计算房源热度特征而热度特征参与了模型输入。解决办法是特征计算只使用训练截止日之前的数据测试期间的行为人工屏蔽。具体定义上我是在生成特征时把event_time test_start_date作为条件确保特征和 label 完全隔离。5.5 推荐结果同质化越推荐越窄用户流失警告协同过滤有一个天然的缺陷所有相似用户会被推荐到同一个头部房源池个性化只是“听起来很美”。如果你的推荐结果前 20 名里超过一半是同一个商圈的热门盘用户很快就会失去兴趣。我的解法是在排序阶段加一个“多样性惩罚项”候选房源与用户最近 10 次浏览房源的属性相似度超过一个阈值时得分乘以 0.8 的折扣强制系统不要把同一类房子连续推给用户。实测这个加权调整后推荐列表的覆盖率提升 30%用户点击位置从第二位提升到第一位整体点击率上涨明显。写在最后这套系统从零到一跑通我最真实的体会是工程能力比算法更能决定线上效果。ALS 是个公开发表多年的算法没什么秘密可言真正的差距在数据清洗、特征工程、任务调度和问题排查这些脏活累活上。你愿意在这上面下功夫哪怕用最简单的内容推荐也能做出不错的效果懒得处理数据质量再花哨的模型上线也是一剂安慰剂。最后分享一个小技巧不要忽略时间衰减。用户的行为是随时间衰减的一个用户 30 天前浏览过的房源跟他昨天刚关注的房源在模型里的比重应该完全不同。给行为数据引入时间衰减权重是我这套系统里人工干预收益最明显的一处改动。往后的扩展方向可以加大在线反馈的实时权重或者尝试用图神经网络替代矩阵分解但无论如何先把数据底座和离线链路打磨扎实比追新模型有价值得多。