#研发解决方案介绍#Recsys-Evaluate（推荐评测）

郑昀 基于刘金鑫文档 最后更新于2014/12/1

关键词： recsys 、推荐评测、Evaluation of Recommender System、piwik、flume、kafka、storm、redis、mysql

---

本文档适用人员：研发

 

推荐系统可不只仅是围着推荐算法打转

  先明确一下，咱们属于工业领域。不少在学术论文里行之有效的新特奇算法，在工业界是行不通的。当年咱们作语义聚合时，分词、聚类、类似性计算、实体词识 别、情感分析等领域最终还都采用了工业界十几年前乃至于几十年前就流行的成熟算法。若是算法不能决定命运，那什么是关键呢？

   算法+规则库+人工干预（整理语料、标识、调参数等），大都是脏活儿累活儿。

  或者叫， 特征+算法+人工干预，用 特征缩小数据范围或降维。

  我在2009年曾经 写道：

在语义的世界里，能够近似地说：万事万物都是特征提取。 你只要找到特征，事情就好办。……

……你指望毕其功于一役吗？天然语言处理的真实应用里是很难有什么场景找到一个通吃特征的。都是一层一层特征叠加的。 

一层特征去掉一部分垃圾数据。如此反复，终成正果。注意方法论。

   梁斌在2012年微博说道：

统计粗且糙，乃大锤。规则细而精，乃小锤。先大场后细棋。

  规则库怎么来的？ 得建设一些方便观测的外围系统，才能发现特征、创建规则、调整参数、观察效果。因此与此相似，作了推荐服务后，就须要推荐效果评测了。

 

推荐评测应用场景

  电商推荐场景下有很是明确的指标：

1. 推荐位展现次数、商品投放次数；
2. 推荐位展现点击率、商品投放点击率；
3. 最重要的是下单转化率和成单转化率（或叫支付转化率）这两个硬指标。

  那么推荐评测系统应具有的功能有：

1. 实时（至少是近乎实时）统计几个展现性指标
• 区分网站端和移动客户端的推荐展现效果
• 进一步区分不一样客户端，如 iOS 和 Android
• 图
2. 数据概览
• 图
3. 按推荐位类型或推荐算法概括各类指标
• 看了又看
• 浏览过该商品的用户购买了
• 您可能对如下商品感兴趣（猜你喜欢）
• 商品周边商品（注：只能是本地生活服务类商品）
• 签到弹窗推荐
• 商品附近门店
• 门店周边美食
• 附近吃喝玩乐
• 图
• ……
4. 常见的评测推荐效果的两种实验方法
1. 离线试验：
• 作法：从日志系统中取得用户的行为数据，而后将数据集分红训练数据和测试数据，好比80%的训练数据和20%的测试数据（还能够交叉验证），而后在训练数据集上训练用户的兴趣模型，在测试集上进行测试
• 优势：它不须要实际用户的交互
• 缺点：离线实验只能评测一个很狭窄的数据集切面，主要是关于算法预测或者评估的准确性
• 目的：提早过滤掉性能较差的算法
2. AB测试：
• 作法：经过必定的规则把用户随机分红几组，并对不一样组的用户采用不一样的推荐算法，这样的话可以比较公平地得到不一样算法在实际在线时的一些性能指标
• 图
5. 推荐服务接口测试界面
• 暴露出来，让咱们手工就能够提交，看看效果

 

推荐评测技术选型

  说到实时日志聚合和处理，还得是 flume+kafka+storm，因此技术选型是：

   Piwik+Flume+Kafka+Storm+Redis+MySQL

 

推荐评测数据流转流程

1. 数据上报：——Piwik
1. 主站自己部署了 开源流量统计系统 Piwik，因此在网页的各类推荐位上按规则埋点便可
• 实例：“浏览过该商品的用户还购买了”推荐栏第一位商品的a元素增长了wwe属性：wwe="t:goods,w:rec,id:ae45c145d1045c9d51c270c066018685,rec:101_01_103"
2. 浏览器彻底加载完成后， Piwik JavaScript 会向服务器端发送埋点数据
3. Piwik 服务器端收到后，写磁盘日志文件
2. 数据采集：——Flume
1. Piwik 集群的每一台服务器上都部署了 Flume Agent
2. agent 会向 推荐数据收集 Flume 集群 推送日志，譬如配置为每增长一行日志就推送，或每5分钟推送一次
3. 手机客户端的埋点日志则存放在无线服务器端的 MySQL 中，因此咱们用脚本每分钟读取一次数据，放到 flume 的监控目录下
3. 数据接入：——Kafka
1. 因为数据采集速度和数据处理速度不必定匹配，所以添加一个消息中间件 Linkedin Kafka 做为缓冲
2. 数据流转方式为 Flume Source-->Flume Channel-->Flume Sink，那么咱们写一个 Kafka Sink 做为消息生产者，将 sink 从 channel 里接收到的日志数据发送给消息消费者
3. 图
4. 流式计算：——Storm
1. Storm 负责对采集到的数据进行实时计算
2. Storm Spout 负责从外部系统不间断地读取数据，并组装成 tuple 发射出去，tuple 被发射后在 Topology 中传播
3. 因此咱们要写一个 Kafka Spout 做为消息消费者拉日志数据
4. 再写些 Storm Bolt 处理数据
• ，一个Topology的结构示意图
5. 数据输出：——Redis
1. Storm Bolt 实时分析数据以后，将统计结果写入 Redis
6. 数据统计：——MySQL
1. 评测系统实时数据直接从 Redis 中读取，并查询主站数据库追踪成单状况，同步到 MySQL 中，做为报表展现数据源

  简而言之，数据按以下方式流转：

1. piwik javascript
2. piwik servers
3. flume agent
4. 自定义 kafka sink
5. 自定义 kafka spout
6. 自定义 storm bolt
7. redis
8. 评测系统计算
9. mysql
10. 评测系统报表展现

 

Flume+Kafka+Storm常见问题

  虽然咱们的实时流量统计和推荐评测系统均采用了 flume+kafka+storm 方案，但要注意这个方案也有一些小坑。下面摘录一些 第三方的结论：

• 如 果配置为每新增一条日志就采集，那么 flume 到 kafka 的实时数据可能会因为单条过快，形成 storm spout 消费 kafka 消息速率跟不上。延时能够是数据发射到 stream 中后进行 hbase 的计算操做引发的（注：hbase 的性能确实堪忧，不适合这种实时数据处理，尤为是加了较多索引以后）；
• 可参考的一个数据：storm 单条流水线的处理能力大约为 20000 tupe/s (每一个tuple大小为1000字节)；
• tuple 过多，会因为 kafka 的 message 须要 new String() 进行获取，会报 gc 的异常；
• tuple 在 stream 中的大量堆积，形成超时自动回调 fail() 的函数；
• 能够进行多 tuple 结构的优化，把多个 log 打包成一个 tuple
• 就通常状况而言，单条发射能扛得住

 

Kafka Sink 消息生产者代码片断

KafkaSink.java

import kafka.javaapi.producer.Producer; …… public class KafkaSink extends AbstractSink implements Configurable { ……  private Producer<String, byte[]> producer; ……  @Override  public Status process() throws EventDeliveryException {   Channel channel = getChannel();   Transaction tx = channel.getTransaction();   try {    tx.begin();    Event e = channel.take();    if (e == null) {     tx.rollback();     return Status.BACKOFF;    }    producer.send(new KeyedMessage<String, byte[]>(topic, e.getBody()));    tx.commit();     return Status.READY;   } catch (Exception e) {

 

Kafka Spout 消息消费者代码片断

spout 有多个，咱们挑 kafka spout 看下。

KafkaSpout.java

public abstract class KafkaSpout implements IRichSpout { ……  @Override  public void activate() { ……    for (final KafkaStream<byte[], byte[]> stream : streamList) {     executor.submit(new Runnable() {      @Override      public void run() {       ConsumerIterator<byte[], byte[]> iterator = stream.iterator();       while (iterator.hasNext()) {        if (spoutPending.get() <= 0) {         sleep(1000);         continue;        }        MessageAndMetadata<byte[], byte[]> next = iterator.next();        byte[] message = next.message();        List<Object> tuple = null;        try {         tuple = generateTuple(message);        } catch (Exception e) {         e.printStackTrace();        }        if (tuple == null || tuple.size() != outputFieldsLength) {         continue;        }        collector.emit(tuple);        spoutPending.decrementAndGet();       }      }

 

Storm Bolt 代码片断

有多个自定义 bolt，挑一个看下。

EvaluateBolt.java

public class EvaluateBolt extends BaseBasicBolt { ……  @Override  public void execute(Tuple input, BasicOutputCollector collector) { ……   if (LogWebsiteSpout.PAGE_EVENT_BROWSE.equals(event)) {    if (LogWebsiteSpout.PAGE_TYPE_GOODS.equals(pageType)) {     incrBaseStatistics(baseKeyMap, BROWSE_ALL, 1);    } else if (LogWebsiteSpout.PAGE_TYPE_PAY1.equals(pageType)) {     incrBaseStatistics(baseKeyMap, ORDER_ALL, 1);    }    String recDisplay = input.getStringByField(LogWebsiteSpout.FIELD_REC_DISPLAY);    recDisplayStatistics(recDisplay, time, pageType, baseKeyMap);   } else if (LogWebsiteSpout.PAGE_EVENT_CLICK.equals(event)) {    String recType = input.getStringByField(LogWebsiteSpout.FIELD_REC_TYPE);

 

评测指标定义：

• 投放点击率：推荐浏览量/推荐商品投放量
• 展示点击率：推荐浏览量/推荐位展示次数
• 推荐展现率:：推荐位展现次数/总浏览量
• 推荐浏览量：经由推荐产生的浏览量
• 推荐商品投放量：推荐位投放的推荐商品数量（如：用户浏览A商品，那在浏览或购买推荐位产生的推荐商品为5个，则推荐商品投放量+5）
• 推荐位展示次数：若是推荐位有推荐商品并展现，计数+1

 

-over-