首页 > 代码库 > 推荐系统

推荐系统

1.缘起

糖豆作为国内最大的广场舞平台，全网的MAU已经超过4000万，每月PGC和UCG生产的视频个数已经超过15万个，每月用户观看的视频也超过100万个。然而之前糖豆APP首页主要还是依赖内容编辑手工推荐来发现内容，每天的推荐量也是几十个而已。明显可见千人一面的内容分发效率比较低下，继而我们于2016年12月初，启动了糖豆推荐系统的设计以及开发，目前截止到2017年1月初，已经完成第一期推荐系统的开发与评估。推荐项目立项伊始，我撰写了一篇整体架构与设计，本文和架构一文在部分内容有所重复，本文主要专注阐述推荐系统的开发、实现以及评估的细节。

推荐系统的目的也可以简单总结成为以下两点：

根据用户个人兴趣分发内容，为生产者和消费者打造更加合理的流量分发体系。
提高用户观看时长，从而进一步到达提升产品留存。

可以看到核心评估目标是用户的观看时长，相对直接易理解。当然评估过程，我们遵循数据科学的评估体系，衡量了包括多种优化目标(RMSE,P@K,AUC/ROC,覆盖率等等)的指标。同时还根据AB测试，评估了整体推荐模块的CTR,播放时长等多项业务统计指标。

2.架构

相信自从Netfix公布他们的推荐架构之后[1]，后续的推荐系统基本都会按照在线(online),近线(near line),离线(off line)三个部分来构建。虽然划分成三个模块，本质是推荐算法迭代时间窗口问题，根据用户行为数据，构建一个持续进化的系统。

糖豆推荐系统架构基本也是按照三个模块来构建。限于人力和时间，第一期主要实现了离线部分。架构图如下：

推荐系统架构1.0

整个系统架构主要由数据、算法、策略、评估和服务层组成，相对清晰明了。

数据层，主要数据来源包括用户行为日志以及数据库。我们在16年10月份~11月份，对整个日志收容、分析和挖掘流程做了改造。
- 收容：同时将日志离线和在线的pipeline彻底分离。
- 解析：原本基于MR的ETL全部改为Spark任务，在集群机器数量不变情况下，整体效率基本提高了两倍以上，Spark具备很好的取代MR的潜力。
- 挖掘：Spark MLib集成了多种机器学习算法，原有基于Mahout的算法基本可以替代实现。
算法层架构一图中，黑字部分是我们实现了的算法，蓝字部分都是计划中但未实现的算法。
策略层：
- 融合算法，主要包括以下三种，目前我们同时使用了级联联合以及混合融合。
- 业务过滤，目前暂时没做。
- 推荐排序，目前的排序对用户隐式反馈行为(包括播放时长、下载、收藏等指标)做线性加权以及归一化处理，得到一个0~5分之间的评分，作为LFM的数据集，通过模型得到预测的打分，最后按照视频打分以及视频创建时间做倒序排序。后续我们会引入学习排序(LTR)算法，来持续改进推荐结果排序质量。LTR，包括PointWise,PairWise,ListWise三类算法。预期未来先使用PointWise类别的算法。

3.算法实现

推荐系统算法在过去几十年有非常长足的发展和应用，总结下来基本包括基于内容、基于邻域，基于矩阵分解等类型。

基于邻域：核心思想是，为用户推荐与之属性、行为相似的物品。邻域就是兴趣相似的数学表达。它包括UserCF和ItemCF，基础研究深入，在性能、可解释性上效果都不错，所以应用也十分广泛。
基于矩阵分解：也就是隐语义模型，在文本挖掘范围首先被提出。矩阵分解是一系列复杂算法(LSM,LSI,LDA,Topic Model)的数学基础。它包括特征值分解、奇异值分解等，有具体计算方法包括SVD,Funk-SVD,ALS,SVD++等。

3.1 LFM

隐语义模型其核心思想是通过潜在特征联系用户和物品,根据用户行为统计的自动聚类。LFM模型能够划分出多维度、软性、不同权重的分类。它通过以下数学公式来表达用户对物品的兴趣，由两个低秩的矩阵来近似表达原有高阶矩阵。

矩阵分解

可以看到从矩阵计算问题，转化成优化问题。优化目标的数学形式化：

优化形式化

这个形式化问题有多种解法，包括SVD,ALS等。Spark提供了包括mlib里的ALS，以及graphx里的SVD++。

3.1.1 ALS(最小交替二乘法)

ALS将矩阵计算转化成为一个最优化函数问题,通过最小化误差的平方和计算最佳函数匹配。ALS在每次迭代期间，一个因子矩阵保持恒定，而另一个使用最小二乘法求解。同样在求解另一因子矩阵，保持新求解的因子矩阵固定不变。

Spark ALS的实现，每次迭代过程了为了减少通讯消耗，只会传输两个因子矩阵(用户、物品)之一参与计算。这个实现是通过预计算矩阵的元数据，得到一个meta矩阵。这样就可以在用户和物品block之间只传输一组特征向量，来更新计算。

ALS

优点，不受到用户和数据质量影响。全局性求解，单一模型效果最好。
缺点，增量更新缓慢。

3.1.2 Spark实现

spark mlib实现了ALS算法，调用比较简单，稍微麻烦的是调参和评估。贴段python代码，注释比较详细了。

##初始化sparksession(spark 2.0以上引入)
spark = SparkSession.builder.master(‘yarn-client‘).appName(‘recy_als_model:‘+inUVMDate).config(‘spark.sql.warehouse.dir‘, ‘/user/hive/warehouse‘).enableHiveSupport().getOrCreate()
#读入用户视频评分全量表
rateSql =  "select * from da.recy_als_data_uvm where dt=‘"+inUVMDate+"‘"
#spark 读hive表
rating = spark.sql(rateSql)
#分割训练集和测试集,0.8,0.2
(training, test) = rating.randomSplit([0.8, 0.2])
#ALS模型参数
ranks = [8, 10]
lambdas = [0.01,0.05, 0.1]
numIters = [20]
bestModel = None
bestValidationRmse = float("inf")
bestRank = 0
bestLambda = -1.0
bestNumIter = -1
#调参
for rank, lmbda, numIter in itertools.product(ranks, lambdas, numIters):
    als = ALS(rank=rank,maxIter=numIter, regParam=lmbda, userCol="f_diu", itemCol="f_vid", ratingCol="f_rating", nonnegative=True)
    model = als.fit(training)
    #!!注意是随机取样，使用测试集评估模型，通过RMSE来评估模型。由于测试集中可能有模型中没出现过的user,那就会有预测值为nan。drop即可
    predictions = model.transform(test).dropna(‘any‘)
    evaluator = RegressionEvaluator(metricName="rmse", labelCol="f_rating",
                                    predictionCol="prediction")
    validationRmse = evaluator.evaluate(predictions)
    print "RMSE (validation) = %f for the model trained with " % validationRmse +           "rank = %d, lambda = %.1f, and numIter = %d." % (rank, lmbda, numIter)
    if (validationRmse < bestValidationRmse):
        bestModel = model
        bestValidationRmse = validationRmse
        bestRank = rank
        bestLambda = lmbda
        bestNumIter = numIter
# evaluate the best model on the test set
print "The best model was trained with rank = %d and lambda = %.1f, " % (bestRank, bestLambda)   + "and numIter = %d, and its RMSE on the test set is %f." % (bestNumIter, bestValidationRmse)
#保存预测结果
predictions = bestModel.transform(rating).dropna(‘any‘)
predictPath =  "hdfs://Ucluster/olap/da/recy_als_predict/"+inUVMDate+"/"
predictions.repartition(200).write.mode(‘overwrite‘).save(predictPath, format="parquet")

spark.stop()

spark ml库在逐步取代mlib库，我们使用了ml，上面代码片段需要引入pyspark.ml相关的类。

3.1.3 候选集问题

我们训练模型数据量基本在10亿量级，我们计算集群总共16台8核，24G的datanode，训练时间大概30分钟。按照我们用户和物品规模，如果直接使用模型预测推荐结果，候选集规模在万亿级别，是集群无法承受的。所有需要对预测的候选集做过滤，目前采用三种过滤方法。

看过的作者。将用户过去30天看过的作者的作品作为候选集。这个做法合理清晰，但是存在所谓的“信息茧房”问题，也就是说容易出现多样性不足。
看过的相似的视频。根据ItemCF算法得到相似的视频。将过去看过30天的Top10的类似视频当作候选集。
看过的相似的标签的视频。将用户看过的视频相同类型标签的视频作为候选集。依赖专家知识，在具体到我们的舞蹈视频上，我们编辑提供的标签只能覆盖极少的视频。由于这种做法倾向于PGC作者，在测试后期不再使用。

3.2 ItemCF(基于物品的协同过滤)

基于物品的协同过滤算法是目前应用最广泛的推荐算法，由亚马逊提出[2]，核心思想给用户推荐那些和他们之前喜欢物品相似的物品。相似度是基于用户对物品的行为来计算的，而非物品本身的属性。

3.2.1 算法原理

基于物品的协同过滤算法主要分为以下两步：

计算物品之间的相似度
根据物品的相似度和用户历史行为给用户生成推荐列表

核心是计算物品之间的相似度，我们使用余弦相似度。

余弦相似度

该算法惩罚了热门物品的权重，减轻热门视频和大量视频相似的可能性。

3.2.2 Spark实现

我们基于spark sql实现了ItemCF，贴一段

   spark = SparkSession.builder.master(‘yarn-client‘).appName(‘recy_icf_similarity:‘+inDate).config(‘spark.sql.warehouse.dir‘, ‘/user/hive/warehouse‘).enableHiveSupport().getOrCreate()
   #指定spark 分区数
   spark.sql("SET spark.sql.shuffle.partitions=2000")
   spark.sql("drop table if exists da.recy_icf_similarity_mid ")
   spark.sql("create table da.recy_icf_similarity_mid as select a.vid vid_1 , b.vid vid_2 , a.num num_1, b.num num_2, count(1) num_12 from da.recy_icf_similarity_pre a join da.recy_icf_similarity_pre b on (a.diu=b.diu) where a.vid<b.vid group by a.vid, b.vid, a.num, b.num")
   #计算余弦相似度
   similarSql = " select vid_1, vid_2, num_12/sqrt(num_1*num_2) similarity from da.recy_icf_similarity_mid"
   similarDF = spark.sql(similarSql)
   similarDF.printSchema()
   # 保存结果
   similarDF.repartition(300).write.mode(‘overwrite‘).save(similarDir, format="parquet")
   spark.stop()

3.3 抄底策略

抄底策略其实是一个冷启动的问题，策略也非常多。

近期热门item、新item。
编辑精选。
新品类上线。
同城热门。

我们目前只生效了热门策略，采用了Hack News的热门算法作为抄底策略，如下图：

热门算法

P表示视频观看次数。
T表示距离视频发布时间(单位为小时),加上2是为了防止最新的视频导致分母过小。
G表示"重力因子"(gravityth power),即为视频衰减系数。

热门算法衰减系数

我们根据实验结果，确定了G的取值。该算法同时保证了视频的热门程度和新鲜度。sql代码如下：

SELECT vid,title,createtime,hits_total,(if( hits_total>=1, hits_total - 1,hits_total)/power((TIMESTAMPDIFF(hour,createtime,now())+2),1.8)) as sc FROM `video` WHERE date(createtime) >=NOW() - INTERVAL 3 DAY ORDER BY `sc`

3.4 算法融合

融合策略主要包括以下三类，当然还有ensemble相关的方法：

加权融合(Weight Merge):根据经验值对不同算法赋给不同的权重，对各个算法产生的候选集按照给定的权重进行加权，然后再按照权重排序
级联融合(Cascade Merge): 优先采用效果好的算法，当产生的候选集大小不足以满足目标值时，再使用效果次好的算法。
混合融合(Mix Merge): 不同的算法按照不同的比例产生一定量的候选集，然后叠加产生最终总的候选集。

我们主要在候选集上使用了mix merge，在结果产出时，采用了cascade merge合并LFM和ItemCF的结果。

4.服务实现

4.1 AB分桶服务

根据用户diu，使用crc32 hash函数对用户取余，分别赋予AB两个类型。客户端拿到abtag后根据服务端数据流实现展示和数据埋点。

4.2 推荐服务

个性化推荐系统服务会在app首页打开后被调用，具体服务流程步骤如下：

通过用户DIU获取推荐模型导出的数据列表
判断推荐的数据列表是否为空
推荐的数据列表如果不为空，则执行5
推荐的数据列表如果为空，则获取抄底的推荐列表，然后执行5
从推荐的数据列表中过滤点目前首页已经展现的视频
根据推荐的分数和视频创建时间，将列表进行排序
返回结果

流程图

4.3 存储选型

推荐系统每天出一次推荐结果，因此推荐结果需要按天区分, 同时需要按diu来快速查询，可以采用的存储有hbase，redis等键值对数据库，mongodb等文档型数据库，或者
mysql等传统关系型数据库

hbase 键值对存储，存储量大，查询速度快，稳定性取决于集群是否高可用，如高可用，可优先选择
redis 键值对存储，存储量较大，热数据基于内存存储，查询速度快，可以考虑，不过当每个人的推荐结果N较大时，要考虑存储大小
mongodb 文档型数据库，存储量大，热数据同样存储在内存，索引速度接近于redis，结构化，易维护，可以考虑
mysql 关系型数据库，存储量较大，基于文件索引机制，查询速度较上述存储来说，理论值较低，可以作为备选。

每个用户的推荐数N=60，存储占用180g，决定采用hbase 根据rowkey字段做索引，当我们指定diu和date时，会快速返回rowkey在该范围内的结果。

5.效果评估

5.1 离线评估

采用融合多维度用户行为数据线性转换成显式反馈评分。由于采用了多维度数据，算法模型效果大幅提升，结果如下：

RMSE从4.1提升到1.0。(Netfix大赛冠军大概在0.8左右)
P@K从0.6705提升到0.938。
预测覆盖率为99%,推荐覆盖率为90%。

5.2 A/B测试

猜你喜欢模块已经在官方渠道测试将近三周，展现形式如下图：

6.改进与展望

第一期开发的时间相对较短，人力也非常不足，期间还有很多数据分析、挖掘工作需要兼顾，整体工作相对简单。未来第二期，主要精力集中在近线和在线的模块开发，以及学习排序。

推荐系统

声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉：投诉/举报工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。

联系
我们

首页 > 代码库 > 推荐系统

推荐系统

推荐系统

1.缘起

2.架构

3.算法实现

3.1 LFM

3.1.1 ALS(最小交替二乘法)

3.1.2 Spark实现

3.1.3 候选集问题

3.2 ItemCF(基于物品的协同过滤)

3.2.1 算法原理

3.2.2 Spark实现

3.3 抄底策略

3.4 算法融合

4.服务实现

4.1 AB分桶服务

4.2 推荐服务

4.3 存储选型

5.效果评估

5.1 离线评估

5.2 A/B测试

6.改进与展望

看完仍有疑问？有类似问题直接问程序猿