[译]使用Google Cloud计算引擎和机器学习算法实现产品推荐

摘要：经过一段时间的说句搜集，当具备一定的数据量时，你就可以用通过机器学习算法来执行一些有用的分析并产生一些有价值的推荐了。

翻译自 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上获得包括安装教程和源代码在内的完整内容。