摘要：經過一段時間的說句搜集，當具備一定的數據量時，你就可以用通過機器學習算法來執行一些有用的分析并產生一些有價值的推薦了。

翻譯自?Google Cloud Platform 原文標題：Using Machine Learning on Compute Engine to Make Product Recommendations 原文地址：https://cloud.google.com/solu...

在一個網絡商店中，你可以使用谷歌云平臺來創建一個可拓展的、高效可用的服務來向用戶推薦相關的商品。

網購平臺的競爭從沒有像今天這樣激烈過，雖然顧客們在不同的供應商之間花費越來越的錢，但是對于單個零售商而言顧客們話費的錢卻在變少。與此同時，單筆購物的消費也在變少，一部分的原因也是因為這種競爭只需要簡簡單單的一次點擊。向顧客提供相關商品的推薦能夠有效的將潛在顧客變為購買力并提高訂單的平均價值。

通過閱讀這篇文章，你能夠搭建起一個基本的環境用于支撐一個基礎的推薦引擎，你也可以根據自己的需求不斷的調整和完善它。在云平臺上面運行推薦殷勤能夠給開發者提供一個靈活、可拓展的解決方案。

在這篇文章中，你將可以了解到一個真實的房地產租賃公司是如何計算相關推薦并向訪問他們網站的用戶推薦這些產品的。

Samantha正在尋找一幢房子來度過她的假期。她在一家度假網站上面注冊過賬號并且曾經在這個網站上面購買過幾次獨家套餐。Sam想要找到根據自己的性格和品味推薦的房子。系統應該已經知道了她的口味，顯然，根據她的過往訂單，她喜歡的是豪華型的房子，因此，系統也應該向她推薦一些類似的房間。

為了向用戶進行推薦，不管是讓用戶實時瀏覽到還是通過email告知用戶，有以下幾件事情是一定要做的。首先，如果你對用戶的品味和愛好知之甚少，那么你可能只能根據商品的一些屬性多帶帶的進行推薦，但是你的系統一定要有能夠從用戶那兒學習的能力，也就是說能夠從用戶那里手機他們的喜好和行為。經過一段時間的說句搜集，當具備一定的數據量時，你就可以用通過機器學習算法來執行一些有用的分析并產生一些有價值的推薦了。為了讓其他用戶的輸入也能夠改善推薦的結果，推薦系統還需要能夠周期性的進行重新訓練。這篇文章中介紹的主要是已經有足夠數據量能夠進行很好的積極學習訓練的情況下的推薦系統。

一個典型的推薦引擎一般將數據經過以下這四步的處理：

這種系統的結構可以通過一下這張圖表來表示：

每一步都可以進行定制以達到一些特殊的需求，這樣的系統由這些部分組成：

一個可拓展的前端用于記錄與用戶的交互以此采集數據

可以被機器學習平臺訪問到的永久存儲引擎。裝載數據到存儲容器中也包括了一些步驟，比如導入/導出（import- export?）數據和對數據進行一些必要的變形（transformation）

一個機器學習平臺用于分析已有的數據集并產生推薦

可以被前端訪問到的存儲容器，可以是實時的也可以不是，由需求中對推薦時間的要求來決定

為了在速度、簡便性、成本控制和精確度之間的平衡，這篇文章選擇了Google App Engine,?Google Cloud SQL, 和?在?Google Compute Engine?上使用?bdutil運行的Apache Spark?.

App Engine?能夠輕松地處理每秒處理數萬次的請求。不管你是用于創建網站還是用來將數據存儲進后端的存儲容器，App Engine都能夠讓你在很短的時間之內將代碼發布到生產環境中。

Cloud SQL對于開發者而言也非常的簡單。Cloud SQL能夠拓展至32核、208GB內存的虛擬機并且能夠拓展容量至10TB并實現每GB30IOPS和數以千計的并發連接。這樣的性能對于這篇文章的中的例子來說是綽綽有余的，而對于真實環境中的大數據推薦引擎來說，Cloud SQL也提供了能夠直接和Spark交互的特性。

Spark提供了比典型Hadoop引擎更好的性能，在使用?Spark MLlib的情況下，Spark的速度可以比后者快10到100倍。你可以在幾分鐘內分析數以億計的數據，這也增加了推薦系統的敏捷性，使得管理員能夠更加頻繁的運行推薦算法。Spark同時也趨向于使用更加簡化的程序模型，帶來了更加簡單的API使用體驗和更加靈活的語言特性。Spark在調節計算機內存使用的同時也盡可能的減少了硬盤的讀寫頻率，與此同時，Spark也在努力的簡化I/O操作。這個解決方案中，使用Compute Engine來支撐分析所使用的基礎設施，Compute Engine通過其按時、按需求計費的方式盡可能的降低了進行分析所帶來的成本。

以下的這幅流程圖和之前的結構圖是一樣的，但展示了每一步所使用到的技術：

一個推薦系統能夠通過用戶潛在的行為或者明確的輸入來收集用戶相關的數據

行為數據的采集是非常容易的，因為你可以保存用戶活動的各種日志。采集這類數據也是非常簡單直接的因為它不需要用戶其他任何的操作，畢竟他們已經在使用這個應用了。但這個手段的負面之處在于這些數據非常難以分析。舉個例子，從用戶不太感興趣的日志中過去出他們可能感興趣的內容就是一件非常笨重的事情。你可以通過這個列子來看一看使用日志來進行潛在數據分析的例子Real-time Log Analysis using Fluentd and BigQuery

直接輸入的數據相對而言更加難以采集，因為這需要用戶做一些額外的操作，比如寫一條評價。處于各種各樣的原因，用戶可能不太想提供這樣的信息。但如果能夠理解用戶的行為，這樣的結果就十分明確了

你能夠向算法提供的數據集越大，那么你的推薦表現就會越好。這也就以為著，任意的一個推薦系統都有可能很快速的成為一個大數據項目。

你用于創建推薦系統的數據結構能夠幫助你決定你所要使用的存儲引擎。你可以選擇使用NoSQL數據庫、標準的SQL數據庫、甚至其他的一些對象存儲。所有的這些選項都是可行的，這取決于你是在分析用戶的輸入還是行為同時其他的一些因素比如執行的簡便性、這些存儲引擎能夠管理的數據容量，與其他環境的兼容性以及可移植性等等。

當存儲用戶的費率(rating)或者事件(events)時，一個可拓展和可管理的數據庫能夠簡化操作所需要的任務數量并能夠幫助我們把重心放在推薦系統上。?Cloud SQL滿足了所有的這些需求，并且能夠使從Spark直接導入數據變得非常簡單。

以下的實例代碼展示了Cloud SQL數據表的結構。Accommodation表表示了房產的評分，Rating表表示的是用戶對每一具體房產所給出的評分

CREATE TABLE Accommodation
(
  id varchar(255),
  title varchar(255),
  location varchar(255),
  price int,
  rooms int,
  rating float,
  type varchar(255),
  PRIMARY KEY (ID)
);

CREATE TABLE Rating
(
  userId varchar(255),
  accoId varchar(255),
  rating int,
  PRIMARY KEY(accoId, userId),
  FOREIGN KEY (accoId)
    REFERENCES Accommodation(id)
);

Spark能夠從不同的來源中獲得數據，比如Hadoop或者Cloud Storage.本文通過使用?Java Database Connectivity (JDBC) connector直接從Cloud SQL中獲得數據。由于Spark的任務是并行的，因此這個接口必須對所有實例可訪問。

在設計分析算法的時候往往需要充分理解應用的要求，這些要求包括

推薦算法的及時性：程序給出推薦的結果需要多長的時間

對于數據的過濾方法：程序是僅僅根據用戶自己的口味還是包括其他用戶的想法，又或者是邏輯上與這個產品相匹配的

在分析數據時首要考慮的因素就是你的應用需要多長的時間將推薦的結果展示給用戶。如果你需要立即展示你的推薦結果，比如當用戶在瀏覽一個產品的時候，那么相對于向用戶發送包含推薦信息的郵件你需要一個更多靈活性的分析算法。

實時系統可以在數據產生時就進行處理。這種類型的系統往往包括了一些能夠處理和分析數據流的工具。一個實時操作系統需要給出一個及時所見即所得的推薦

批處理推薦需要你能夠周期性的處理數據。這一手段意味著為了分析相關性，足夠的數據需要被產生，比如每日銷量。批處理系統適用于在晚些時候發送郵件推薦這種情況

近實時分析需要你迅速的獲取數據這樣就能夠每隔幾分鐘甚至幾秒鐘刷新分析的數據。一個近實時分析系統適用于在同一個瀏覽會話期間產生推薦的情況。

一個推薦系統可以歸入以上三種及時性標簽的任意一種但是，對于一個在線銷售系統而言，你需要考慮一些結余近實時和批處理之間的情況，這取決于應用能夠獲取的流量和用戶輸入的情況。運行分析的平臺可以直接從數據存儲的平臺開始工作也可以基于一個周期性轉存數據的平臺。

搭建一個推薦系統的核心組件就是過濾，最常用的手段包括

基于內容的推薦（Content-based）：一個受歡迎的被推薦的商品和用戶瀏覽或者喜歡過的有相同的屬性

基于集群的推薦（Cluster）：被推薦的商品總是一起出現，不管其他的用戶做了什么

協同過濾推薦（Collaborative）:喜歡該件商品的其他用戶也喜歡的被推薦的商品

雖然云平臺可以支持任意的一種方法，但是本文主要關注采用協同過濾方法的推薦，這一方法可以通過使用Apache Spark來被執行。想了解更多的關于基于內容和基于集群的推薦算法，可以訪問?appendix

協同過濾算法確保你能夠抽象化商品的屬性并且基于用戶的口味進行預測。這種過濾的輸出基于這樣的一種假設：喜歡過統一商品的不同的兩個用戶現在可能會喜歡同樣的東西

你可以將評分或者互動的數據重新演繹為一個矩陣，商品和用戶各占一個維度。協同過濾方法嘗試針對一個明確的“用戶-產品”對去預測矩陣中缺少的部分。下圖中的兩個矩陣很相似，但是第二個矩陣是根據第一個矩陣中將存在數據的部分替換為1，不存在數據的部分替換為0而預測的。結果矩陣是一個真值表用1表示用戶與產品之間存在聯系。

這里用兩種不同的手段去使用協同過濾方法：

Memory-based filtering：計算產品或者用戶之間的相似度

Model-based filtering?：嘗試去學習用戶與產品之間交互的深層模式

本文使用model-based方法，基于用戶已經訂購過的商品

本文中所用的所有分析手段都可以通過?PySpark獲得，這個接口為Spark程序開發提供了一個Python的封裝。你也可以使用Scala或者Java開發，具體請看?Spark的開發文檔

Spark MLlib使用?Alternating Least Squares (ALS)算法來訓練模型。你可以使用一下幾種參數的多種組合來獲得方差和偏差之間的最好結果

Rank:引導用戶給出評分的未知因素的數量。這些因素可能包含看比如年齡、性別或者所在的地區。在一定的范圍內，rank值越高，那么推薦的效果就會越好。在內存和CPU允許的情況下，從5開始，每一次增加5知道推薦的改善率（improvement rate）放緩會是一個很好的手段

Lambda:一個用來防止過度（overfiting）擬合的正規化參數,由高方差和低偏差值所代表。方差表示通過多次運行理論上的正確值和實際運行結果的波動情況。偏差值則表示你所得到的預測結果與真實結果之間的差距。過度擬合發生在模型在訓練集上能夠非常好的運行但是在實際的測試集上卻不能表現良好。lambda值越高，過度擬合的情況就會越少但是誤差值也會越大。在測試過程中0.01，1和10都是很好的選擇

下面的圖標展示了方差和誤差之間的關系。靶心表示的是算法想要預測的結果。

Iteration：訓練算法所要運行的次數。在這個例子中，你將會運行5，10和20次迭代并使用不同的rank和lambda組合

下面的代碼展示了如何在Spark中開始一個ALS模型訓練

from pyspark.mllib.recommendation import ALS
model = ALS.train(training, rank = 10, iterations = 5, lambda_=0.01)

使用ALS算法的協同過濾器基于三個不同的數據集：

訓練集：包括已知結果的數據。這個數據集會獲得最好的結果。在這篇文章中，它包含的是用戶給出的評分數據

驗證集：包括的數據能夠幫助訓練器去選取合適的參數組合并選擇最好的模型

測試集：包括被用于評估訓練所得到的最好模型的數據。測試集相當于在真實環境中使用這個推薦算法的模擬

為了找到最好的模型，你需要去計算所基于驗證集合模型規模的均方根誤差（root-mean-square error）。均方根誤差越小，模型也就越好

為了讓用戶能夠簡單快速地獲取結果，你需要把結果裝載進可以根據需求被查詢的數據庫。再安利一次，Cloud SQL是一個非常好的選擇。從Spark 1.4開始，你可以使用PySpark將預測結果直接寫進數據庫。

Recommendation?表的結構大概像這樣：

CREATE TABLE Recommendation
(
  userId varchar(255),
  accoId varchar(255),
  prediction float,
  PRIMARY KEY(userId, accoId),
  FOREIGN KEY (accoId)
  REFERENCES Accommodation(id)
);

這一部分將展示訓練模型的一些代碼

Spark SQL的上下文能夠讓你輕松地連接到一個Cloud SQL實例通過JDBC連接器。數據以DataFrame的形式加載

pyspark/app_collaborative.py

jdbcDriver = "com.mysql.jdbc.Driver"
jdbcUrl    = "jdbc:mysql://%s:3306/%s?user=%s&password=%s" % (CLOUDSQL_INSTANCE_IP, CLOUDSQL_DB_NAME, CLOUDSQL_USER, CLOUDSQL_PWD)
dfAccos = sqlContext.load(source="jdbc", driver=jdbcDriver, url=jdbcUrl, dbtable=TABLE_ITEMS)
dfRates = sqlContext.load(source="jdbc", driver=jdbcDriver, url=jdbcUrl, dbtable=TABLE_RATINGS)

Spark使用了一種稱為?Resilient?Distributed?Dataset (RDD)的概念，這一概念使得在并行情況下處理元素變得便利.RDDs是創建于持續存儲的只讀數據集，他們能夠在內存中被處理，因此他們非常適合迭代處理。

回顧如何獲得最好的預測模型，你需要將你的數據分為三個不同的數據集。下面的代碼使用了一個工具函數幫助你隨機的將非重復的數據以60%、20%、20%的比例分開

pyspark/app_collaborative.py

rddTraining, rddValidating, rddTesting = dfRates.rdd.randomSplit([6,2,2])

注意
在創建Rating表的時候以下面的順序創建是非常重要的：accoId,?userId,?rating.ALS需要根于已經定義好的產品-用戶對來進行預測。如果你不是這樣創建的，你可以選擇修改你的數據庫或者使用RDD上的map方法來更新列的順序

回顧一下，當使用ALS方法的時候，系統需要使用rank、regularization和iteration參數來找到最合適的模型。由于評分存在，因此train方法獲得的結果就必須和驗證集進行比較。你需要確認用戶的口味也一起包含在了訓練集之中

pyspark/app_collaborative.py

for cRank, cRegul, cIter in itertools.product(ranks, reguls, iters):
  model = ALS.train(rddTraining, cRank, cIter, float(cRegul))
  dist = howFarAreWe(model, rddValidating, nbValidating)
  if dist < finalDist:
    print("Best so far:%f" % dist)
    finalModel = model
    finalRank  = cRank
    finalRegul = cRegul
    finalIter  = cIter
    finalDist  = dist

注意
howFarAreWe方法僅使用用戶-產品對在驗證集上使用模型來預測評分

pyspark/app_collaborative.py

def howFarAreWe(model, against, sizeAgainst):
  # Ignore the rating column  
  againstNoRatings = against.map(lambda x: (int(x[0]), int(x[1])) )

  # Keep the rating to compare against
  againstWiRatings = against.map(lambda x: ((int(x[0]),int(x[1])), int(x[2])) )

  # Make a prediction and map it for later comparison
  # The map has to be ((user,product), rating) not ((product,user), rating)
  predictions = model.predictAll(againstNoRatings).map(lambda p: ( (p[0],p[1]), p[2]) )

  # Returns the pairs (prediction, rating)
  predictionsAndRatings = predictions.join(againstWiRatings).values()

  # Returns the variance
  return sqrt(predictionsAndRatings.map(lambda s: (s[0] - s[1]) ** 2).reduce(add) / float(sizeAgainst))

現在你已經有一個可以給出充分推薦理由的模型了，你可以使用它來根據用戶的口味和其他與他口味相同用戶的評分預測這個用戶最可能感興趣的東西了。在這一步中，你能夠看到之前提到過的舉證映射（matrix- mapping）

pyspark/app_collaborative.py

 Build our model with the best found values
# Rating, Rank, Iteration, Regulation
model = ALS.train(rddTraining, BEST_RANK, BEST_ITERATION, BEST_REGULATION)

# Calculate all predictions
predictions = model.predictAll(pairsPotential).map(lambda p: (str(p[0]), str(p[1]), float(p[2])))

# Take the top 5 ones
topPredictions = predictions.takeOrdered(5, key=lambda x: -x[2])
print(topPredictions)

schema = StructType([StructField("userId", StringType(), True), StructField("accoId", StringType(), True), StructField("prediction", FloatType(), True)])

dfToSave = sqlContext.createDataFrame(topPredictions, schema)
dfToSave.write.jdbc(url=jdbcUrl, table=TABLE_RECOMMENDATIONS, mode="overwrite")

既然你已經獲得了所有預測值的列表，那么你就可以在Cloud SQl里存儲前十個結果了這樣系統就可以為用戶提供一些推薦了。比如，使用這些預測結果的一個很好的時間就是當用戶登錄進網頁的時候。

pyspark/app_collaborative.py

dfToSave = sqlContext.createDataFrame(topPredictions, schema)
dfToSave.write.jdbc(url=jdbcUrl, table=TABLE_RECOMMENDATIONS, mode="overwrite")

想要運行這個解決方法，你可以一步步按找這個頁面?GitHub page的介紹操作。通過這些指導，你將能夠為用戶計算和展示推薦信息。

最終的SQL代碼將從數據庫中取出排名最高的推薦信息并把它展示在Samantha的歡迎頁面上

查詢語句運行在Cloud Platform控制臺或者Mysql的客戶端時，會返回類似于下圖的結果

在網頁中，同樣的查詢語句會強化歡迎頁面并且增加訪問者成為消費者的可能性

這看上去與Sam和在上文中描述的系統已知的Sam喜歡的東西十分相似。

你可以在?GitHub上獲得包括安裝教程和源代碼在內的完整內容。