Collective的Spark ML经验分享：读者模型

irobot

Collective的Spark ML经验分享：读者模型

作者 孙镜涛 发布于 2015年11月20日

【编者的话】Collective成立于2005年，其总部位于纽约，是一家从事数字广告业务的公司。 该公司的数字广告业务非常依赖于机器学习和预测模型，对于特定的用户在特定的时间应该投放什么样的广告完全是由实时或者离线的机器学习模型决定的。本文来自Databricks的技术博客，Eugene Zhulenev分享了自己在Collective公司从事机器学习和读者模型工作的经验。

   Collective公司有很多使用机器学习的项目，这些项目可以统称为读者模型，因为这些项目都是基于用户的浏览历史、行为数据等因素预测读者转化、点击率等信息的。在机器学习库的选择上，Collective公司内部新开发的大部分项目都是基于Spark和Spark MLLib的，对于一些被大家广泛使用而Spark并不具备的工具和类库Collective还专门创建了一个扩展库Spark Ext。在本文中，Eugene Zhulenev介绍了如何使用Spark Ext和Spark ML两个类库基于地理位置信息和浏览历史数据来预测用户转化。
预测数据
      预测数据包含两种数据集，虽然这些数据都是使用虚拟的数据生成器生成的，但是它们与数字广告所使用的真实数据非常相似。这两类数据分别是：
  用户的浏览历史日志

1. Cookie          | Site          | Impressions  
   
2. --------------- |-------------- | -------------
   
3. wKgQaV0lHZanDrp | live.com      | 24
   
4. wKgQaV0lHZanDrp | pinterest.com | 21
   
5. rfTZLbQDwbu5mXV | wikipedia.org | 14
   
6. rfTZLbQDwbu5mXV | live.com      | 1
   
7. rfTZLbQDwbu5mXV | amazon.com    | 1
   
8. r1CSY234HTYdvE3 | youtube.com   | 10

复制代码

  经纬度地理位置日志

1. Cookie          | Lat     | Lng       | Impressions
   
2. --------------- |---------| --------- | ------------
   
3. wKgQaV0lHZanDrp | 34.8454 | 77.009742 | 13
   
4. wKgQaV0lHZanDrp | 31.8657 | 114.66142 | 1
   
5. rfTZLbQDwbu5mXV | 41.1428 | 74.039600 | 20
   
6. rfTZLbQDwbu5mXV | 36.6151 | 119.22396 | 4
   
7. r1CSY234HTYdvE3 | 42.6732 | 73.454185 | 4
   
8. r1CSY234HTYdvE3 | 35.6317 | 120.55839 | 5
   
9. 20ep6ddsVckCmFy | 42.3448 | 70.730607 | 21
   
10. 20ep6ddsVckCmFy | 29.8979 | 117.51683 | 1

复制代码

  转换预测数据
  正如上面所展示的，预测数据是长格式，对于每一个cookie与之相关的记录有多条，通常情况下，这种格式并不适合于机器学习算法，需要将其转换成“主键——特征向量”的形式。
  Gather转换程序     
      受到了R语音 tidyr和reshape2包的启发，Collective将每一个键对应的值的长数据框（long DataFrame）转换成一个宽数据框（wide DataFrame），如果某个键对应多个值就应用聚合函数。

1. val gather = new Gather()
   
2.       .setPrimaryKeyCols("cookie")
   
3.       .setKeyCol("site")
   
4.       .setValueCol("impressions")
   
5.       .setValueAgg("sum")         //通过key对impression的值求和
   
6.       .setOutputCol("sites")
   
7. val gatheredSites = gather.transform(siteLog)

复制代码

  转换后的结果

1. Cookie           | Sites
   
2. -----------------|----------------------------------------------
   
3. wKgQaV0lHZanDrp  | [
   
4.                  |  { site: live.com, impressions: 24.0 },
   
5.                  |  { site: pinterest.com, impressions: 21.0 }
   
6.                  | ]
   
7. rfTZLbQDwbu5mXV  | [
   
8.                  |  { site: wikipedia.org, impressions: 14.0 },
   
9.                  |  { site: live.com, impressions: 1.0 },
   
10.                  |  { site: amazon.com, impressions: 1.0 }
    
11.                  | ]

复制代码

  Google S2几何单元Id转换程序
  Google S2几何类库是一个球面几何类库，该库非常适合于操作球面（通常是地球）上的区域和索引地理数据，它会为地球上的每一个区域分配一个唯一的单元Id。
  为了将经纬度信息转换成键值对的形式，Eugene Zhulenev结合使用了S2类库和Gather，转换后数据的键值是S2的单元Id。

1. // Transform lat/lon into S2 Cell Id
   
2. val s2Transformer = new S2CellTransformer()
   
3.   .setLevel(5)
   
4.   .setCellCol("s2_cell")
   
5. 
   
6. // Gather S2 CellId log
   
7. val gatherS2Cells = new Gather()
   
8.   .setPrimaryKeyCols("cookie")
   
9.   .setKeyCol("s2_cell")
   
10.   .setValueCol("impressions")
    
11.   .setOutputCol("s2_cells")
    
12. 
    
13. val gatheredCells = gatherS2Cells.transform(s2Transformer.transform(geoDf))

复制代码

  转换后的结果

1. Cookie           | S2 Cells
   
2. -----------------|----------------------------------------------
   
3. wKgQaV0lHZanDrp  | [
   
4.                  |  { s2_cell: d5dgds, impressions: 5.0 },
   
5.                  |  { s2_cell: b8dsgd, impressions: 1.0 }
   
6.                  | ]
   
7. rfTZLbQDwbu5mXV  | [
   
8.                  |  { s2_cell: d5dgds, impressions: 12.0 },
   
9.                  |  { s2_cell: b8dsgd, impressions: 3.0 },
   
10.                  |  { s2_cell: g7aeg3, impressions: 5.0 }
    
11.                  | ]

复制代码

  生成特征向量
      虽然Gather程序将与某个cookie相关的所有信息都组织到了一行中，变成了键值对的形式，但是这种形式依然不能作为机器学习算法的输入。为了能够训练一个模型，预测数据需要表示成double类型的向量。
  Gather 编码程序     
      使用虚拟变量对明确的键值对进行编码。

1. // Encode S2 Cell data
   
2. val encodeS2Cells = new GatherEncoder()
   
3.   .setInputCol("s2_cells")
   
4.   .setOutputCol("s2_cells_f")
   
5.   .setKeyCol("s2_cell")
   
6.   .setValueCol("impressions")
   
7.   .setCover(0.95) // dimensionality reduction

复制代码

  原始数据

1. Cookie           | S2 Cells
   
2. -----------------|----------------------------------------------
   
3. wKgQaV0lHZanDrp  | [
   
4.                  |  { s2_cell: d5dgds, impressions: 5.0 },
   
5.                  |  { s2_cell: b8dsgd, impressions: 1.0 }
   
6.                  | ]
   
7. rfTZLbQDwbu5mXV  | [
   
8.                  |  { s2_cell: d5dgds, impressions: 12.0 },
   
9.                  |  { s2_cell: g7aeg3, impressions: 5.0 }
   
10.                  | ]

复制代码

  转换后的结果

1. Cookie           | S2 Cells Features
   
2. -----------------|------------------------
   
3. wKgQaV0lHZanDrp  | [ 5.0  ,  1.0 , 0   ]
   
4. rfTZLbQDwbu5mXV  | [ 12.0 ,  0   , 5.0 ]

复制代码

      对于转换后的结果，用户还可以根据场景选择性地使用顶部转换进行降维。首先计算不同用户每个特征的值，然后根据特征值进行降序排序，最后从结果列表中选择最上面那些数值总和占所有用户总和的百分比超过某个阈值（例如，选择最上面覆盖99%用户的那些网站）的数据作为最终的分类值。
  Spark ML 管道
        Spark ML 管道是Spark MLLib的一个新的高层API。一个真正的ML管道通常会包含数据预处理、特征提取、模型拟合和验证几个阶段。例如，文本文档的分类可能会涉及到文本分割与清理、特征提取、使用交叉验证训练分类模型这几步。在使用Spark ML时，用户能够将一个ML管道拆分成多个独立的阶段，然后可以在一个单独的管道中将他们组合到一起，最后使用交叉验证和参数网格运行该管道从而找到最佳参数集合。
  使用Spark ML管道将它们组合到一起

1. // Encode site data
   
2. val encodeSites = new GatherEncoder()
   
3.   .setInputCol("sites")
   
4.   .setOutputCol("sites_f")
   
5.   .setKeyCol("site")
   
6.   .setValueCol("impressions")
   
7. 
   
8. // Encode S2 Cell data
   
9. val encodeS2Cells = new GatherEncoder()
   
10.   .setInputCol("s2_cells")
    
11.   .setOutputCol("s2_cells_f")
    
12.   .setKeyCol("s2_cell")
    
13.   .setValueCol("impressions")
    
14.   .setCover(0.95)
    
15. 
    
16. // Assemble feature vectors together
    
17. val assemble = new VectorAssembler()
    
18.   .setInputCols(Array("sites_f", "s2_cells_f"))
    
19.   .setOutputCol("features")
    
20. 
    
21. // Build logistic regression
    
22. val lr = new LogisticRegression()
    
23.   .setFeaturesCol("features")
    
24.   .setLabelCol("response")
    
25.   .setProbabilityCol("probability")
    
26. 
    
27. // Define pipeline with 4 stages
    
28. val pipeline = new Pipeline()
    
29.   .setStages(Array(encodeSites, encodeS2Cells, assemble, lr))
    
30. 
    
31. val evaluator = new BinaryClassificationEvaluator()
    
32.   .setLabelCol(Response.response)
    
33. 
    
34. val cROSsValidator = new CrossValidator()
    
35.   .setEstimator(pipeline)
    
36.   .setEvaluator(evaluator)
    
37. 
    
38. val paramGrid = new ParamGridBuilder()
    
39.   .addGrid(lr.elasticNetParam, Array(0.1, 0.5))
    
40.   .build()
    
41. 
    
42. crossValidator.setEstimatorParamMaps(paramGrid)
    
43. crossValidator.setNumFolds(2)
    
44. 
    
45. println(s"Train model on train set")
    
46. val cvModel = crossValidator.fit(trainSet)

复制代码

  结论
      Spark ML API让机器学习变得更加容易。同时，用户还可以通过Spark Ext创建自定义的转换/估计，并对这些自定义的内容进行组装使其成为更大管道中的一部分，此外这些程序还能够很容易地在多个项目中共享和重用。如果想要查看本示例的代码，可以点击这里。