Universal Recommender (UR) 是 ActionML 公司基于 PredictionIO 开发的协同过滤引擎，使用 Correlated Cross-Occurrence (CCO) 算法，可以应用与个性化推荐、物品相关推荐、购物车推荐、基于业务逻辑的推荐等。
COO算法是协同过滤算法的一种，然而一般的协同过滤只能针对单一行为，CCO算法可以计算交叉行为下的协同关联。例如：它不仅可以通过用户的浏览行为来告诉你 “浏览了内容A的人可能会浏览内容B” ,它还能结合用户的浏览行为和用户的广告点击行为来告诉你 “点击了广告A的人可能会浏览内容F”。关于COO的详细介绍可参考下面文章：基于 CCO 的协同过滤推荐，Multi-domain predictive AI or how to make one thing predict another。
PredictionIO 与 UR 安装
参考 PredictionIO 与 UR 的官方文档安装，需要注意的是 UR 需要使用 Elasticsearch 来保存结果，现时（PredictionIO 0.12.1）默认支持的 Elasticsearch 版本为 5.。最简单的方式就是下载tar包然后解压到 PredictionIO 的 vonders/elasticsearch-5..* 目录，然后在pio-env.sh文件进行配置。
配置完成后可以在 ur 的目录下运行以下命令进行测试：
./examples/integration-test
此命令会使用 examples/handmade-engine.json 的配置，data/sample-handmade-data.txt 中的数据，导入数据，build模型，deploy模型，使用预设数据进行测试并且与预设结果进行对比。
DASE实现
DataSource 只接受 user-item 类型，事件类型可以有多种，如测试数据中的 purchase、view、category-pref，其中第一个为主事件，其余为次要事件。可以使用 $set 请求定义 item 属性。DataSource 读取事件日志，按事件名聚合事件；读取 item 信息，并且将 Properties 信息附加到 item 上；然后返回事件、Item相关的RDD。
Preparator 根据 minEventsPerUser 的配置过滤掉主事件过少的用户，然后过滤掉没有主事件的用户以及其次要事件。因为 UR 算法需要用到 mahout，Preparator 需要将事件 RDD 转换为 org.apache.mahout.sparkbindings.indexeddataset.IndexedDatasetSpark 供算法使用。IndexedDatasetSpark 包含 userID 的 BiDictionary，itemID 的 BiDictionary 以及 经过 BiDictionary key转换的 user-item 矩阵。
Serving 与 RecommendationEngine 分别为用户请求的响应处理以及引擎的构造描述，较常规没有太多特殊内容。
URAlgorithm 根据 URAlgorithmParams 配置训练模型，保存，然后做预测。train 过程计算主事件矩阵与自身以及个次要事件的 Log Likelihood Ratio（LLR） 矩阵，然后保存到 ElasticSearch。predict 过程将用户的输入转换为对 ElasticSearch 的搜索请求，此为模型核心。
模型参数保存：历遍各 LLR 矩阵中的每一行，按降序排序每行的 releate_item，并且构造成 (itemID, Map(actionName -> JArray(vector[releate_item]))) ，然后通过 groupAll 操作，将每个 item 的事件内容与属性内容集合到一起，每个条目均得到类似以下格式内容：
(Ipad-retina,Map(defaultRank -> JDouble(3.0), expires -> JString(2018-06-15T17:09:41.677740+00:00), countries -> JArray(List(JString(United States), JString(Estados Unidos Mexicanos))), date -> JString(2018-06-13T17:09:41.677740+00:00), category-pref -> JArray(List(JString(tablets))), categories -> JArray(List(JString(Tablets), JString(Electronics), JString(Apple))), available -> JString(2018-06-11T17:09:41.677740+00:00), purchase -> JArray(List(JString(Iphone 6), JString(Iphone 4))), popRank -> JDouble(2.0), view -> JArray(List(JString(Soap)))))
然后创建 Elasticsearch 表索引，最后将 RDD 保存到 Elasticsearch，我们可以先看看数据在 Elasticsearch 中的存在方式，这样会更加清晰：
GET /urindex_1529048147127/_search
{
"_index": "urindex_1529048147127",
"_type": "items",
"_id": "Iphone 4",
"_score": 1,
"_source": {
"defaultRank": 5,
"expires": "2018-06-18T10:45:41.677+08:00",
"countries": [
"United States",
"Canada",
"Estados Unidos Mexicanos"
],
"id": "Iphone 4",
"date": "2018-06-16T10:45:41.677+08:00",
"category-pref": [
"tablets"
],
"categories": [
"Phones",
"Electronics",
"Apple"
],
"available": "2018-06-14T10:45:41.677+08:00",
"purchase": [
"Ipad-retina",
"Iphone 6"
],
"popRank": 4,
"view": [
"Soap",
"Tablets"
]
}
}
很明显这是一个基于产品的推荐系统，主事件为 purchase ，所以直观的解释就是 “购买了 Iphone 4 的用户还购买了 Ipad-retina，Iphone 6；购买了 Iphone 4 的用户浏览了 Soap，Tablets；Iphone 4 属于 Phones，Electronics，Apple 分类……”
UR 模型的 predict 过程实际就是根据查询输入以及用户的历史事件（从Event Storage中查出），构造 ES 的 json query。
模型定义查询结构如下：
case class Query(
user: Option[String] = None, // must be a user or item id
userBias: Option[Float] = None, // default: whatever is in algorithm params or 1
item: Option[String] = None, // must be a user or item id
itemBias: Option[Float] = None, // default: whatever is in algorithm params or 1
itemSet: Option[List[String]] = None, // item-set query, shpping cart for instance.
itemSetBias: Option[Float] = None, // default: whatever is in algorithm params or 1
fields: Option[List[Field]] = None, // default: whatever is in algorithm params or None
currentDate: Option[String] = None, // if used will override dateRange filter, currentDate must lie between the item's
// expireDateName value and availableDateName value, all are ISO 8601 dates
dateRange: Option[DateRange] = None, // optional before and after filter applied to a date field
blacklistItems: Option[List[String]] = None, // default: whatever is in algorithm params or None
returnSelf: Option[Boolean] = None, // means for an item query should the item itself be returned, defaults
// to what is in the algorithm params or false
num: Option[Int] = None, // default: whatever is in algorithm params, which itself has a default--probably 20
from: Option[Int] = None, // paginate from this position return "num"
eventNames: Option[List[String]], // names used to ID all user actions
withRanks: Option[Boolean] = None) // Add to ItemScore rank fields values, default false
extends Serializable
然后转换为类似下面的请求：
{
"size": 20
"query": {
"bool": {
"should": [
{
"terms": {
"rate": ["0", "67", "4"]
}
},
{
"terms": {
"buy": ["0", "32"],
"boost": 2
}
},
{ // categorical boosts
"terms": {
"category": ["cat1"],
"boost": 1.05
}
}
],
"must": [ // categorical filters
{
"terms": {
"category": ["cat1"],
"boost": 0
}
},
{
"must_not": [//blacklisted items
{
"ids": {
"values": ["items-id1", "item-id2", ...]
}
},
{
"constant_score": {// date in query must fall between the expire and available dates of an item
"filter": {
"range": {
"availabledate": {
"lte": "2015-08-30T12:24:41-07:00"
}
}
},
"boost": 0
}
},
{
"constant_score": {// date range filter in query must be between these item property values
"filter": {
"range" : {
"expiredate" : {
"gte": "2015-08-15T11:28:45.114-07:00"
"lt": "2015-08-20T11:28:45.114-07:00"
}
}
}, "boost": 0
}
},
{
"constant_score": { // this orders popular items for backfill
"filter": {
"match_all": {}
},
"boost": 0.000001 // must have as least a small number to be boostable
}
}
}
}
}
Universal Recommender 虽然原理比较简单，不过它巧妙地利用 Elasticsearch 将业务规则整合进来的方式确实可以给人带来启发。
以上仅为个人总结笔记，能力所限可能有比较多错误，欢迎交流指正 yplam(at)yplam.com