推03，最最最簡單的推薦系統是什麼樣的 | 附Spark實踐案例

文·HCY崇遠

接前面這篇《推02，就算是非技術人員也都有必要了解的一些推薦系統常識》，之前的開篇01/02，其實都是以理論、場景化，概念進行鋪墊的，讓大伙兒大概知道推薦系統是怎麼回事，從這篇開始，照顧一下技術的童鞋，我們開始回歸到技術層面，並且附上代碼案例（見後面部分）。當然，依然是入門級，高高高手可以繞路。

01 什麼是最最最簡單的推薦機制

如標題，既然是「最最最簡單」的推薦系統，其實也不能說是推薦系統，之前也說了，系統是一個複合的完整系統，所以這裡說推薦機制可能會更恰當些。結合之前大致陳述的一些推薦機制，最最最簡單的推薦機制，無疑是基於主體屬性相似或者相關推薦了，連個性化都說不上，鐵定最最最簡單了。

說到這，說不定有些人不願意幹了，既然如此簡陋的推薦機制，不看也罷。BUT，真的不要小看基於內容相似的推薦，存在即合理。我們在進行信息獲取時，其實本身就是具有一定識別能力的，這意味著我們最終選擇查看的信息都是經過我們大腦大致思考的結論，這意味著這些信息是有參考價值的。

並且，在很多時候，我們是需要同類信息進行我們獲取到的信息進行補全的，完善我們對目標信息的獲取程度。基於這個考量，基於內容屬性的推薦其實是說的過去的。

別不服，我們來上例子，還是以前面文章那個騰訊視頻的推薦場景圖為例。

圖是在使用騰訊視頻觀看視頻時，我親手截推薦欄位的內容，補充一下背景（很重要，請注意）：

1 我當時觀看的是應該是《藍色星球第二季》紀錄片

2 我經常在騰訊視頻上看的一般是大片，並且一般是國外的

3 由於是VIP賬號，梓塵兄也經常用這個賬號看動畫片，諸如《小豬佩奇》之類的

4 在騰訊視頻上，紀錄片中，我只看過《地球脈動》和《藍色星球》，並且，我真不是紀錄片的愛好者，只是喜歡這兩部而已

基於上面我提供的個人行為數據，再結合看這批推薦列表，不難發現，上面有很多的紀錄片，你覺得跟我們當時正在瀏覽的內容有沒有關係？或者你認為我行為記錄中很多紀錄片的記錄？又或者是我是紀錄片的狂熱者，導致了騰訊視頻給我猛推紀錄片。

所以，連騰訊視頻都會考慮基於當前瀏覽內容的屬性進行推薦（並且是大範圍），你還覺得這種做法十分之LOW嗎？當然你也可以認為騰訊視頻推的不準，瞎J吧推，也是可以的，我也認同，不是非常准（哈哈，《地球脈動》所有我都看過了，還給我瞎推，上面給推的沒幾個有慾望去點的，給騰訊視頻推薦的開發兄弟們打臉了，不好意思）。

我只想表達的是，這種簡單的推薦機制，在整個推薦系統中真的是不可缺少的部分，很多時候簡單並不代表無效，類似上面這種情況，我可以舉出太多有名有姓的實際案例來，說多了沒意義，所以，咱繼續。

02 其實並沒有這麼簡單

從直接的推薦機制來看，整個實現流程看著真的很簡單，但是在實際的操作過程中，還是有一些東西值得探討以及注意的。

第一、首先是，相似計算的過程

之前文章有大致提到過，相似或者相關計算還是有很多可以選擇的，他們每一種都有各自的特點以及適應性。以相似計算中使用最多的歐式距離與餘弦相似為例，專業點的說法就是餘弦夾角可以有效規避個體相同認知中不同程度的差異表現，更注重維度之間的差異，而不注重數值上的差異，而歐式距離則是對個體異常數值會比較敏感。

這意味著，在我們需要區分異常樣本時，使用距離計算會更恰當，聚個栗子，比如電商領域中高價值與低價值用戶的區分，其實我們核心是想把他們的差異性拉大的，得以體現出對比，這個時候使用餘弦就是不合理的。

在回歸到距離上說，市面上除了歐式距離，還有好幾種距離度量，諸如馬氏、曼哈頓距離等等，其實其度量側重都是不一樣的，我們需要結合實際的場景去使用。還有更偏向於相關度量的皮爾森相關係數等。

第二、需要解決相似計算中，計算矩陣過大的問題

按照標準流程，假設有1萬個物品，則對於每個物品來說，需要與其他物品計算與其的相似度或者相關度，然後再排個序，取TopN形成自身的待推薦列表。那麼，簡單的數學題來了10000*10000=10000萬次計算，這顯然是不合理的。

所以，優化這個過程是必然的，關鍵是如何優化。核心思想其實就是初篩嘛！把那些完全沒啥多大鳥關係的直接幹掉，省掉計算相似的過程，節省資源。如何篩選？一個比較常見的做法是，尋找核心關鍵影響因素，保證關鍵因素的相關性。

比如，在實際的生產操作過程中，很多時候會通過關鍵屬性是否基本匹配作為判斷依據，或者直接通過搜索構建進行檢索初篩，把大致相關的先過濾一遍，通過這種簡單而粗暴的方式，其實已經能把大部分相關度很低的候選集給過濾掉，對於整體計算量級來說，計算複雜度直接下降。

第三、多個因子如何權衡影響權重

基於屬性計算相似，從整體上來看，其實一般主體都不止一個屬性，那麼計算相關的時候到底看那個屬性呢？或者說哪些屬性應該佔有更高的權重，哪些因素是次要因素。

還是以上面的騰訊視頻的推薦為例，從結果上來反推相似推薦的部分（當然，實際情況不詳哈，只是推斷而已），顯然當前視頻的類別佔了很大的權重（記錄片），除此之外包括導演啊，一些其他特徵屬性應該也會參考在內的。

回到常規問題，如何確定影響權重是個操作難題。最簡單並且實際上還挺有效的一種方式就是專家評判法，即通過權威經驗來劃定影響因子的權重，還有就是通過標註的樣本進行反向擬合每種因素的佔比權重。除此之外還有一些其他學術上的方法，包括什麼主成分分析法，層次分析法，還有什麼熵權法，其實都是找因子影響能力的主次關係。

最終確定好了影響因素，在實際上線回收到數據之後，依然是需要逐步的進行權重影響調整的，我們可以通過結果的樣本數據，進行LR的回歸擬合，尋找最合適的權重配比。

03 說實踐案例比較實在

說了這麼多理論，不能光說不練，標題上寫著「附Spark案例」，很多人都是沖著這來的呢，前面BB了這麼多屁話，還不見代碼。來，我們這就上正文。

不過不用期待過多，畢竟這只是一個簡單的相似計算的過程而已，所以權當屬性實驗數據以及Spark開發了，高手可以略過了。

一、實驗數據簡介

其實看到這三部分數據的簡介，一些老手估計已經知道是什麼數據了，是的，就是那份有名的電影數據集（MovieLens開放數據），並且取的是完全版的那份，簡直成了推薦系統的標配實驗數據了。

三個文件，其中電影數據集共1萬多個電影基礎數據，評分數據集最大共100萬條評分數據，以及10萬條的用戶對電影的打標籤數據，總大小約為幾百兆，不大，但是用來做實驗玩玩那是相當足夠了。

二、推薦機制邏輯

我們的核心計算邏輯還是內容屬性上的相似嘛，所以核心是看看圍繞電影，有哪些屬性是可以抽取出來的，並且參與計算的。

第一，電影的類別，基於上面騰訊視頻的考慮，其實這個顯然很重要，而電影的類別信息存儲於電影數據集中，並且是一對多的關係，即一個電影可以對應多個類目，我們需要進行切割，由於計算這個維度相似的時候，是多對多的關係，天然的計算相似或者相關的特徵。

第二、電影的播放年份，電影的年份其實有種潛在的關聯關係，舉個例子可以說明，比如說零幾年的電影與現狀的電影風格是不同的，當時間跨度有一定差距時，這個還是蠻明顯的。關於電影的年份數據，從數據樣本可以知道，它隱藏在電影的名字中，寫個正則過濾出來即可。至於說如何計算這個維度的相關，我們可以用兩者差距來算相關，比如年份絕對值越遠，這個值越小，越近甚至是重疊就越大。

第三，電影的標籤，電影本身是沒有標籤屬性的，但它有用戶對他打的標籤信息，所以我們需要進一步處理，把它變成電影的屬性，需要清洗、規整以及處理。標籤本身也是多對多的關係，同樣可以計算相似度，比如歐式或者餘弦。

第四、電影的名稱，名稱上進行尋找關聯性，聽上去很扯，但其實有一定的邏輯在裡頭，比如我在視頻網站搜索「三國」，顯然我期望從名稱上尋找三國相關題材的視頻，他們就是在名稱上建立起關聯關係的，所以，名稱從某種程度上來說，可以體現相關性。在計算相似或者相關方式上，我們可以進行分詞，去除停詞，然後再以詞維度進行餘弦計算。

第五、候選集電影的評分，對於做推薦來說，首先需要保證的推薦的候選集一定是優質的，從這個維度上說，拋開其他因素，那麼就是整體評分高的電影是相對優質的電影。在處理的過程中，由於一個電影對應多個評分，所以，我們需要進行進行歸一計算，最簡單的做法就是計算整體評分的平均值，作為電影的評分數據，評分過低的數據直接從候選集中幹掉，又大大的降低了計算次數。

三、代碼邏輯

Spark2.0之後，不用再構建sparkcontext了，以創建一個複合多功能的SparkSession替代，可以正常的從HDFS讀取文件，也可以從Hive中獲取DataFrame等等。

val sparkSession = SparkSession .builder() .appName("base-content-Recommand") //spark任務名稱 .enableHiveSupport() .getOrCreate()

那三個表可以先load到Hive中，然後spark直接從Hive中讀取，形成DataFrame。

//從hive中，獲取rating評分數據集，最終形成如下格式數據(movie,avg_rate)val movieAvgRate = sparkSession.sql("select movieid,round(avg(rate),1) as avg_rate from tx.tx_ratings group by movieid").rdd.map{ f=> (f.get(0),f.get(1))}//獲取電影的基本屬性數據，包括電影id，名稱，以及genre類別val moviesData = sparkSession.sql("select movieid,title,genre from tx.tx_movies").rdd//獲取電影tags數據，這裡取到所有的電影tagval tagsData = sparkSession.sql("select movieid,tag from tx.tx_tags").rdd

先對tag進行處理，很多tag其實說的是同一個東西，我們需要進行一定程度上的合併，這樣才能讓tag更加的合理（有朋友有意見了，就一個實驗案例而已，搞這麼複雜），舉個簡單例子，blood、bloods、bloody其實都是想說這個電影很血腥暴力，但是不同的人使用的詞不同的（這點大伙兒可以自由查看實驗數據），所以我們需要進行一定程度上的合併。

val tagsStandardizeTmp = tagsStandardize.collect()val tagsSimi = tagsStandardize.map{ f=> var retTag = f._2 if (f._2.toString.split(" ").size == 1) { var simiTmp = "" val tagsTmpStand = tagsStandardizeTmp .filter(_._2.toString.split(" ").size != 1 ) .filter(f._2.toString.size < _._2.toString.size) .sortBy(_._2.toString.size) var x = 0 val loop = new Breaks tagsTmpStand.map{ tagTmp=> val flag = getEditSize(f._2.toString,tagTmp._2.toString) if (flag == 1){ retTag = tagTmp._2 loop.break() } } ((f._1,retTag),1) } else { ((f._1,f._2),1) }}

其中getEditSize是求取，兩個詞的編輯距離的，編輯距離在一定時候，進行合併，具體邏輯見代碼了，不複雜。

def getEditSize(str1:String,str2:String): Int ={ if (str2.size > str1.size){ 0 } else { //計數器 var count = 0 val loop = new Breaks //以較短的str2為中心，進行遍歷，並逐個比較字元 val lengthStr2 = str2.getBytes().length var i = 0 for ( i <- 1 to lengthStr2 ){ if (str2.getBytes()(i) == str1.getBytes()(i)) { //逐個匹配位元組，相等則計數器+1 count += 1 } else { //一旦出現前綴不一致則中斷循環，開始計算重疊度 loop.break() } } //計算重疊度,當前綴重疊度大於等於2/7時，進行字元串合併，從長的往短的合併 if (count.asInstanceOf[Double]/str1.getBytes().size.asInstanceOf[Double] >= (1-0.286)){ 1 }else{ 0 } }}

繼續對tag進行處理，統計tag頻度，取TopN個作為電影對應的tag屬性。

val movieTag = tagsSimi.reduceByKey(_+_).groupBy(k=>k._1._1).map{ f=> (f._1,f._2.map{ ff=> (ff._1._2,ff._2) }.toList.sortBy(_._2).reverse.take(10).toMap)}

接下來處理年齡、年份和名稱，這個會簡單點，進行分詞處理的話，怎麼簡單怎麼來了，直接使用第三方的HanLP進行關鍵詞抽取作為分詞結果，直接屏蔽了停用詞。

val moviesGenresTitleYear = moviesData.map{ f=> val movieid = f.get(0) val title = f.get(1) val genres = f.get(2).toString.split("|").toList.take(10) val titleWorlds = HanLP.extractKeyword(title.toString, 10).toList val year = movieYearRegex.movieYearReg(title.toString) (movieid,(genres,titleWorlds,year))}

取年份的正則函數如下，是個Java寫的精通工具類（Scala和Java混寫，簡直無比美妙）。

package utils;import java.util.regex.Matcher;import java.util.regex.Pattern;/** * Desc: 抽取年份公式 */public class movieYearRegex { private static String moduleType = ".* \(([1-9][0-9][0-9][0-9])\).*"; public static void main(String[] args){ System.out.println(movieYearReg("GoldenEye (1995)")); } public static int movieYearReg(String str){ int retYear = 1994; Pattern patternType = Pattern.compile(moduleType); Matcher matcherType = patternType.matcher(str); while (matcherType.find()) { retYear = Integer.parseInt(matcherType.group(1)); } return retYear; }}

通過join進行數據合併，生成一個以電影id為核心的屬性集合。

val movieContent = movieTag.join(movieAvgRate).join(moviesGenresTitleYear).map{ f=> //(movie,tagList,titleList,year,genreList,rate) (f._1,f._2._1._1,f._2._2._2,f._2._2._3,f._2._2._1,f._2._1._2)}

相似計算開始之前，還記得我們之前說的嗎，可以進行候選集閹割，我們先根據一些規則裁剪一下候選集。

val movieConetentTmp = movieContent.filter(f=>f._6.asInstanceOf[java.math.BigDecimal].doubleValue() < 3.5).collect()

然後真正的開始計算相似，使用餘弦相似度計算，取排序之後的Top20作為推薦列表。

val movieContentBase = movieContent.map{ f=> val currentMoiveId = f._1 val currentTagList = f._2 //[(tag,score)] val currentTitleWorldList = f._3 val currentYear = f._4 val currentGenreList = f._5 val currentRate = f._6.asInstanceOf[java.math.BigDecimal].doubleValue() val recommandMovies = movieConetentTmp.map{ ff=> val tagSimi = getCosTags(currentTagList,ff._2) val titleSimi = getCosList(currentTitleWorldList,ff._3) val genreSimi = getCosList(currentGenreList,ff._5) val yearSimi = getYearSimi(currentYear,ff._4) val rateSimi = getRateSimi(ff._6.asInstanceOf[java.math.BigDecimal].doubleValue()) val score = 0.4*genreSimi + 0.25*tagSimi + 0.1*yearSimi + 0.05*titleSimi + 0.2*rateSimi (ff._1,score) }.toList.sortBy(k=>k._2).reverse.take(20) (currentMoiveId,recommandMovies)}.flatMap(f=>f._2.map(k=>(f._1,k._1,k._2))).map(f=>Row(f._1,f._2,f._3))

最後，將結果存入Hive中，Hive中提前建好結果表。

//我們先進行DataFrame格式化申明val schemaString2 = "movieid movieid_recommand score"val schemaContentBase = StructType(schemaString2.split(" ") .map(fieldName=>StructField(fieldName,if (fieldName.equals("score")) DoubleType else StringType,true)))val movieContentBaseDataFrame = sparkSession.createDataFrame(movieContentBase,schemaContentBase)//將結果存入hive,需要先進行臨時表創建val userTagTmpTableName = "mite_content_base_tmp"val userTagTableName = "mite8.mite_content_base_reco"movieContentBaseDataFrame.registerTempTable(userTagTmpTableName)sparkSession.sql("insert into table " + userTagTableName + " select * from " + userTagTmpTableName)

到這裡，基本大的代碼邏輯就完了，可能還有一些邊邊角角的代碼遺漏了，但不妨礙主幹了。

04 寫在最後

寫到這裡，一篇有業務、有理論、還有代碼的硬文章就出來了，不過在文章中嵌代碼總是很難搞的，想要看整體的代碼，還是得看工程。

想要進一步研究代碼邏輯以及實際跑一跑這個Spark實驗案例的，可以加我，從我這要實驗數據以及完整的代碼文件，不收錢。

當然，如果你是初學者，想要向我了解更多推薦相關的東西，以及代碼邏輯進行詳細的講解，甚至是跑到Spark環境中的整體流程，這個是要收諮詢費的，哈哈，老老實實給我打100大洋的紅包，包服務到家（沒辦法窮），甚至後續的《基於用戶畫像的推薦》、《基於協同的推薦》相關的實踐工程都一併講了，聯繫微信見下。

關於我：

大數據行業半個老鳥，我家梓塵兄的超級小弟，會敲代碼、會寫文章，還會泡奶粉哄小屁孩。

想和我交流的，可以加我個人微信mute88，可以拉你入交流群，但請註明身份and來意~

系列文章：

《推01，你們是不是都覺得自己少了個推薦系統？》

《推02，就算非技術人員也有必要了解的推薦系統常識》

《推03，最最最簡單的推薦系統是什麼樣的 | 附Spark實踐案例》（本文）

歡迎關注數據蟲巢（ID：blogchong），這裡有很多關於大數據的原創文章，如果你覺得文章有用，歡迎轉發，也不介意你打賞一杯深夜寫文的咖啡，謝謝。

推薦閱讀：