糖豆推薦系統第一期開發與評估報告

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

整個系統架構主要由數據、算法、策略、評估和服務層組成，相對清晰明了。

1. 數據層，主要數據來源包括用戶行為日志以及數據庫。我們在16年10月份~11月份，對整個日志收容、分析和挖掘流程做了改造。
   • 收容：同時將日志離線和在線的pipeline徹底分離。
   • 解析：原本基於MR的ETL全部改為Spark任務，在集群機器數量不變情況下，整體效率基本提高了兩倍以上，Spark具備很好的取代MR的潛力。
   • 挖掘：Spark MLib集成了多種機器學習算法，原有基於Mahout的算法基本可以替代實現。
2. 算法層架構一圖中，黑字部分是我們實現了的算法，藍字部分都是計划中但未實現的算法。
3. 策略層：
   • 融合算法，主要包括以下三種，目前我們同時使用了級聯聯合以及混合融合。
   • 業務過濾，目前暫時沒做。
   • 推薦排序，目前的排序對用戶隱式反饋行為(包括播放時長、下載、收藏等指標)做線性加權以及歸一化處理，得到一個0~5分之間的評分，作為LFM的數據集，通過模型得到預測的打分，最后按照視頻打分以及視頻創建時間做倒序排序。后續我們會引入學習排序(LTR)算法，來持續改進推薦結果排序質量。LTR，包括PointWise,PairWise,ListWise三類算法。預期未來先使用PointWise類別的算法。

3.算法實現

推薦系統算法在過去幾十年有非常長足的發展和應用，總結下來基本包括基於內容、基於鄰域，基於矩陣分解等類型。

1. 基於鄰域：核心思想是，為用戶推薦與之屬性、行為相似的物品。鄰域就是興趣相似的數學表達。它包括UserCF和ItemCF，基礎研究深入，在性能、可解釋性上效果都不錯，所以應用也十分廣泛。
2. 基於矩陣分解：也就是隱語義模型，在文本挖掘范圍首先被提出。矩陣分解是一系列復雜算法(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分鍾。按照我們用戶和物品規模，如果直接使用模型預測推薦結果，候選集規模在萬億級別，是集群無法承受的。所有需要對預測的候選集做過濾，目前采用三種過濾方法。

1. 看過的作者。將用戶過去30天看過的作者的作品作為候選集。這個做法合理清晰，但是存在所謂的“信息繭房”問題，也就是說容易出現多樣性不足。
2. 看過的相似的視頻。根據ItemCF算法得到相似的視頻。將過去看過30天的Top10的類似視頻當作候選集。
3. 看過的相似的標簽的視頻。將用戶看過的視頻相同類型標簽的視頻作為候選集。依賴專家知識，在具體到我們的舞蹈視頻上，我們編輯提供的標簽只能覆蓋極少的視頻。由於這種做法傾向於PGC作者，在測試后期不再使用。

3.2 ItemCF(基於物品的協同過濾)

基於物品的協同過濾算法是目前應用最廣泛的推薦算法，由亞馬遜提出[2]，核心思想給用戶推薦那些和他們之前喜歡物品相似的物品。相似度是基於用戶對物品的行為來計算的，而非物品本身的屬性。

3.2.1 算法原理

基於物品的協同過濾算法主要分為以下兩步：

1. 計算物品之間的相似度
2. 根據物品的相似度和用戶歷史行為給用戶生成推薦列表

核心是計算物品之間的相似度，我們使用余弦相似度。

余弦相似度

該算法懲罰了熱門物品的權重，減輕熱門視頻和大量視頻相似的可能性。

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首頁打開后被調用，具體服務流程步驟如下：

1. 通過用戶DIU獲取推薦模型導出的數據列表
2. 判斷推薦的數據列表是否為空
3. 推薦的數據列表如果不為空，則執行5
4. 推薦的數據列表如果為空，則獲取抄底的推薦列表，然后執行5
5. 從推薦的數據列表中過濾點目前首頁已經展現的視頻
6. 根據推薦的分數和視頻創建時間，將列表進行排序
7. 返回結果

流程圖

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.改進與展望

第一期開發的時間相對較短，人力也非常不足，期間還有很多數據分析、挖掘工作需要兼顧，整體工作相對簡單。未來第二期，主要精力集中在近線和在線的模塊開發，以及學習排序。