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)三个部分来构建。虽然划分成三个模块，本质是推荐算法迭代时间窗口问题，根据用户行为数据，构建一个持续进化的系统。
糖豆推荐系统架构基本也是按照三个模块来构建。限于人力和时间，第一期主要实现了离线部分。架构图如下：
整个系统架构主要由数据、算法、策略、评估和服务层组成，相对清晰明了。
- 数据层，主要数据来源包括用户行为日志以及数据库。我们在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之间只传输一组特征向量，来更新计算。
- 优点，不受到用户和数据质量影响。全局性求解，单一模型效果最好。
- 缺点，增量更新缓慢。
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测试
猜你喜欢模块已经在官方渠道测试将近三周，展现形式如下图：
通过AB测试，可以看到首页模块的点击率整体提升了10%，人均观看时长整体提升5%。目前可以看到，猜你喜欢模块效果略优于每日精选。
6.改进与展望
第一期开发的时间相对较短，人力也非常不足，期间还有很多数据分析、挖掘工作需要兼顾，整体工作相对简单。未来第二期，主要精力集中在近线和在线的模块开发，以及学习排序。
