实时推荐引擎设计

一. 系统设计目的
　　<small>根据实时binlog数据和日志流数据实时更新学习模型，做到在线预测</small>

二. 业务场景介绍
　　<small> 目前线上每天有200ｗ左右的订单，而且某些用户可能一天有多次下单，因此用户的实时下单数据对用户做个性化推荐有着重要的意义。而且目前都是T+1的方式进行预测，对于某些高频用户推荐的时效性并不高，需要设计一套实时推荐引擎。</small>

三. 系统设计目标

• <small>实时性：时延能在秒级别</small>
• <small>扩展性强：通用性强，不受限于接入数据格式和算法种类</small>
• <small>可靠性高：服务稳定可靠，即使在流量突增也能应对</small>
• <small>框架简单化，算法模型复杂化：算法模型可任意配置</small>
• <small>特征动态扩展：可根据数据源随意添加新特征并实时反馈到在线预测中</small>
• <small>支持模型ABTest：可对两个不同的模型做在线ABTest，评估效果</small>
• <small>监控：时延，在线预测时间，实时数据流量，模型效果评估</small>
• <small>可维护性：排查问题方便，代码修改简单</small>
• <small>稳定性：即使系统的某一个模块挂掉也不造成服务的停止</small>

四. 系统设计方案
　　<small>系统架构图如下：</small>

实时推荐引擎架构.png

<small>分为四个模块：离线ETL模块，数据聚合模块，学习模块，实时预测模块，配置模块</small>

• 离线ETL模块：
  　　<small>负责实时清洗数据流并存入NoSQL中，NoSQL中的数据是按照主题表去存入的，比喻订单维度表，用户维度表等，每个表按照阶段写入，每个阶段是一个数据块，以供聚合模块增量计算(具体请看第五节)。考虑到数据流的多样性，可能并不能包含维度表的各个特征字段，所以只需要提取主键和相关的特征字段。ETL过程只是负责处理各个不同的数据流并持久化到NoSQL中。</small>

• 数据聚合模块：
  　　<small>定期读取ES模块的数据根据配置模块配置的特征进行提取和计算。该模块能够增量和全量的计算特征并存储到cache中。需要定义新特征时读取存储模块数据全量计算存储到cache；在实时处理过程中只需要增量计算即可。</small>

• 学习模块：
  　　<small>目前有两种方案，如下：
  　　　　1.定期读取配置模块获取最新特征，接着读取cache中数据做学习训练(需要从NoSQL中读取uid,根据uid获取训练集)；
  　　　　2.实时训练，并定期读取配置模块，定期保存训练模型，如果配置发生变化，从头开始训练模型，重新学习，但是需要获取所有用户数据(需要从NoSQL中读取uid,根据uid获取训练集)；
  　　　　评价：方案1模型训练实时性不高，设计简单，便于维护；方案2实时性好，但需要实时获取变更数据(考虑如何保存uid在从cache中取)，要求算法支持增量训练；
  </small>

• 实时预测模块：
  　　<small>定期更新缓存的训练模型，再根据uid读取用户特征数据，在线预测(可以考虑通过消息驱动，而不是定期读取)</small>

• 全局配置模块：
  　　<small>负责配置各个算法的特征，预测类别，特征顺序，为聚合模块、学习模块、在线预测模块提供配置服务</small>
  　　<small>各个模块的交互如下所示：</small>

  
  实时推荐引擎设计.png

五. 各模块具体考虑问题和解决办法

• 离线ETL模块：
  　　<small>问题：</small>
  　　　<small>ａ：针对不同数据流如何做解析？非标准化数据如日志流如何做解析?</small>
  　　　<small>ｂ：解析后的数据格式是怎样的？统一的数据格式如何定义?</small>
  　　　<small>ｃ：对于新数据流是否可以简单快速解析？</small>
  　　　<small>ｄ：对于大量数据如何做到实时解析？是否考虑jstorm？</small>
  　　<small>解决办法：</small>
  　　　<small>ａ：binlog数据结构化，直接通过消费kafka的方式读取指定数据表的数据；日志数据可以指定topic，对每一个topic数据通过定义关键阶段和正则匹配的方式，选取指定数据做解析；考虑使用处理器链的方式来处理每个topic下的数据，如果需要消费其他格式数据只需要定义一个处理器即可。</small>
  　　　<small>ｂ：解析数据格式应该是表名+主键+字段方式，比喻订单格式：(payinfo,orderNo,amount,orderTime)；统一的数据格式为：表名、主键<k,v> 、字段名列表List(<k,v>)</small>
  　　　<small>ｃ：接入新数据流，只需要实现相应的处理器即可，保证能解析成统一的数据格式</small>
  　　　<small>ｄ：可以考虑使用jstrom做到并行处理，每个topology消费一个topic的数据，增加处理器只需要增加一个bolt即可</small>

• ES存储模块：
  　　<small>问题：</small>
  　　　<small>ａ：是否支持按照主键频繁更新并且对查询效率影响不大？</small>
  　　　<small>ｂ：是否支持动态添加字段？</small>
  　　　<small>ｃ：ETL过后的数据如何存储(一个宽表 or 多个维度表)？</small>
  　　　<small>ｄ：根据定义的存储结构是否可以方便的对数据做聚合？提取特征？(如果多个维度表涉及到join)</small>
  　　　<small>ｅ：需要存储大量的用户历史数据，根据以上考虑ES是否真的合适？</small>
  　　<small>解决办法：</small>
  　　　<small>ａ：在QPS较大时ES更新数据会影响查询性能，HBase未调研</small>
  　　　<small>ｂ：ES支持动态添加字段，但没有HBase性能好</small>
  　　　<small>ｃ：如果是ES，多个维度表性能肯定比宽表好，因为每次更新数据就是delete-index操作；如果是HBase，两种设计影响不大</small>
  　　　<small>ｄ：ES和HBase都不支持join，如果跨表做聚合，必须去做关联查询</small>

　　　ｅ：HBase在动态添加字段和频繁更新的情况下或许更好

• 数据聚合模块：
  　　<small>问题：</small>
  　　　<small>ａ：如何做到数据实时增量计算？</small>
  　　　<small>ｂ：是否考虑jstorm去做实时计算？</small>
  　　　<small>ｃ：特征计算：根据数据流计算所有的定义特征然后组装算法特征？ or 直接根据算法配置计算特征？</small>
  　　　<small>ｄ：在数据量较大，特征维度较多的时候如何做到低延迟？</small>
  　　　<small>ｅ：如何解决数据更新导致统计错误的问题？如何解决用户下单链路中数据流延迟(因为数据流分散到不同的topic中)问题？</small>
  　　<small>解决办法：</small>
  　　　<small>ａ：数据阶梯写入原则，统计时候按照阶梯增量统计，两种可行方案：
  　　　　　　　　1.使用时间段将数据分离，比喻19:00_{20:00是一个数据块，20:00}21:00是另一个数据块
  　　　　　　　　2.按照数据量将数据分离，比喻第1000_{1500是一个数据块，1500}2000是另一个数据块</small>
  　　　<small>ｂ：可以考虑使用jstorm做实时处理，按照字段的维度或者特征维度并行执行，比喻提取特征平均订单金额和最大订单金额，可以针对amount字段设置一个bolt，或者直接设置两个bolt(需要考虑算法的特征计算不能保证同时执行完毕的情况)</small>
  　　　<small>ｃ：考虑到特征有复用的情况，根据定义的特征计算然后组装各个算法特征，能大大减少计算量</small>
  　　　<small>ｄ：时延取决于实时增量计算的策略和并行计算的能力。如果数据阶梯划分越细，jstorm并行计算所有特征吞吐越大，则延迟越低。但是需要考虑数据延迟可能导致特征计算错误的问题</small>

　　　 ｅ：目前没有太好的办法，如果不能解决数据流问题，只能在一致性和实时性之间做折衷。比喻延迟计算前几个阶梯的数据，但是会降低实时性，或者直接计算当前阶梯数据，但是不能保证一致性

• cache模块：
  　　<small>问题：</small>
  　　　<small>ａ：如何存储各个算法的特征数据？</small>
  　　　<small>ｂ：是否支持算法特征数据的变化？</small>
  　　　<small>ｃ：是否可以快速的根据算法查询到训练集和根据用户查询某个算法的特征数据以便预测？</small>
  　　<small>解决办法：</small>
  　　　<small>ａ：以uid为key建立map，map中存放用户的特征数据(<k,v>形式)，需要使用的时候根据配置的特征顺序构造特征向量</small>
  　　　<small>ｂ：如果按照a的思路，可以支持特征数据的动态变化</small>
  　　　<small>ｃ：如果按照a的思路，可以支持</small>

• 学习模块：
  　　<small>问题：</small>
  　　　<small>ａ：如何解决算法语言和框架语言的不一致性？</small>
  　　　<small>ｂ：如何隔离各个算法做定期调用？</small>
  　　　<small>ｃ：如何保证数据做算法的增量调用</small>
  　　　<small>ｄ：如何存储训练得到的模型？</small>
  　　<small>解决办法：</small>

*　　　ａ：需要调研 *

<small>ｂ：可以考虑使用调度器，配置调度时间</small>

　　　ｃ：先思考学习系统如何能获取cache中所有数据？(cache不允许做全量扫描)，当前架构是否合理？另外，突然提取新特征，所有用户的新特征如何更新到缓存中？

　　　ｄ：可以存储在本地，每个训练的模型都带有版本号，有没有更好的方式，调研训练后的模型如何存储？

• 全局配置模块：
  　　<small>问题：</small>
  　　　<small>ａ：配置项有哪些？</small>
  　　　<small>ｂ：配置文件存储到哪里？qconfig？mysql(如何做配置的推送)？</small>
  　　　<small>ｃ：如何严格保证特征配置的顺序？</small>
  　　　<small>ｄ：是否需要定义特征的提取规则？如果需要如何定义？</small>
  　　<small>解决办法：</small>
  　　　<small>ａ两种配置：
  　　　　　1.特征配置项：特征名
  　　　　　2.算法配置项包括：算法名、特征名(和cache中一致)、特征顺序、支付方式类别、当前配置版本</small>
  　　　<small>ｂ：配置文件存放到mysql中或者缓存，学习模块定期读取训练好的模型，根据读取的模型版本读取对应版本的特征数据(模型和特征数据配置都缓存起来)，不做消息的推送</small>
  　　　<small>ｃ：通过特征的配置顺序实现</small>
  　　　<small>ｄ：不需要，特征的提取在聚合阶段完成，聚合阶段实现所有的特征计算。</small>
• 实时预测模块：
  　　<small>问题：</small>
  　　　<small>ａ：</small>
  　　<small>解决办法：</small>
  　　　<small>ａ：</small>