1.实时推荐系统与相关工作
1.1 原因
实时计算能够及时捕获用户短时兴趣，同时能够快速反馈分发当前系统的用户兴趣内容。大量实践以及发表的文章都显示了推荐系统实时化，对推荐精准度的提升的有效性和必要性。
1.2 腾讯架构与实现
实时推荐相关工作非常多，腾讯和北大合作的两篇SIGMOD文章是比较实际和详细的实现，采用的计算框架能够支持大规模数据的实时推荐，以下将会分开简述以下两篇文章。
2015年
Huang发表了基于Storm和KV存储的大规模实时推荐系统 (TencentRec: Real-time Stream Recommendation in Practice)
- 实现了一系列经典推荐算法的实时版本
- 实现了数种实时算法提高推荐精度
- 广泛应用于业务有效提高
腾讯采用使用storm原因，支持实时数据流式计算，良好的可扩展性、可容错性，采用简单编程模型。
文章核心包括实时增量计算的ItemCF，以及用户隐式反馈计算、实时剪枝算法、基于用户画像的数据稀疏性策略。应用在多个业务上都有不同程度的提升，最明显的是腾讯视频的全局表现提升高达30%。
全文核心应该是下图六道公式，阐述腾讯如何具体实现的增量itemcf。
文章中的co-rating,其实就是我们常说的user bias. 公式3和4解决了用户隐式反馈问题，细节的计算可以参考2016的文章，实际是一个log函数融合了用户的浏览、点击、分享、购买等行为，转化成rating.
请注意公式4，由于他们定义了corating，实际是将相似度的增量计算从L2范数的计算转化成了L1范数计算.(当Rup取x的时候，y=1/x)。
可扩展的增量计算
2016年
腾讯视频的推荐应用(Real-time Video Recommendation Exploration)
- 实时处理、大规模数据下的准确率和可扩展性。
- 开发了一个基于矩阵分解的大规模在线协同过滤算法，以及一系列的自适应更新策略。
- 通过增加包括视频类别、时间因素影响、用户画像剪枝以及训练等方法，提高实时TopN推荐的精度。
在我们看来，全文核心在于实时计算的数据流转，如下图所示：
基于storm的实时计算topology图:
2. 糖豆的设计与实现
2.1 架构
糖豆整体推荐框架，从离线，近线，在线三套计算流程组合而成。在线流程基于Spark Streaming框架实现，部署在近线集群。 在线推荐框架实时根据用户行为，生成实时推荐列表，从而满足用户瞬时兴趣，提高推荐系统的推荐新鲜度。简单架构图如下:
2.2 基于Spark Streaming的实现
2.2.1. 计算流程
实时计算流程如下图所示:
分解步骤：
- Spark Streaming 读取Kafka，原始日志ETL
- 提取用户隐式反馈，生成候选集tuple (uid,vid)
- 每天凌晨会将离线计算好的ItemCF模型结果集导入Redis。itemcf数据结构是一个similarity vid list。
- 实时维护看过视频set,对看过视频的处理候选集tuple过滤该用户看过的视频
- 实时更新推荐过视频set,候选集tuple过滤当天已经被推荐过的视频
- 候选集写入Redis推荐list
python实现：
if __name__ == "__main__":
print sys.argv
reload(sys)
sys.setdefaultencoding('utf-8')
sc = SparkContext(appName="real_time_etl")
#20秒
ssc = StreamingContext(sc, 15)
brokers = "kafka-servers:9092"
topic = "logstash"
#读取kafka
kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": brokers})
#解析日志、过滤无关数据、读取相似视频
lines = kvs.map(lambda x : readJson(x[1])).filter(lambda x: x is not None).map(lambda x: getTopkfromRedis(x))
#lines.pprint()
#写入推荐结果
lines.foreachRDD(lambda rdd: list2Redis(rdd))
ssc.start()
ssc.awaitTermination()
2.2.2 监控
部署在集群Master节点的监控脚本会每30s扫描一次实时计算代码进程，如果发现进程被failed，会自动拉起实时计算Spark Steaming进程。如果进程拉起失败会触发邮件、短信报警
#! /bin/sh
MOBILE="your phone numbers"
RT_HOME=/home/realtime/recommend.py
DIR=/data/rtdamon
PID_FILE=$DIR/.run/rt-litetl-damon.pid
LOG_FILE=$DIR/.log/rt-litetl-damon.log
t=$(date -d "today" +"%Y-%m-%d %H:%M:%S")
source /etc/profile
echo $PID_FILE $LOG_FILE
if [ -e "$PID_FILE" ];
then
pid=`cat $PID_FILE`
echo $pid
damon_process_exists=`ps v -p $pid | grep "rt-litetl-damon.sh" | grep -v grep|grep -v \<defunct\> `
echo "damon process exists : $process_exists"
if [ -n "$damon_process_exists" ]
then
echo "Process rt-litetl-damon.sh is running! $t" >> $LOG_FILE
exit
fi
fi
pid=$$
echo "$pid" > $PID_FILE
while :
do
process_exists=`ps -ef|grep "$RT_HOME"|grep "spark"|grep -v grep|wc -l`
echo "process exists : $process_exists" >>$LOG_FILE
if [ "$process_exists" == "0" ]; then
/hadoop/spark/bin/spark-submit --master yarn --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.1.0 --py-files /hadoop/user/rt/redis.zip --num-executors 10 --executor-cores 7 --executor-memory 6g /home/realtime/recommend.py>>/data/rtlog/rtrecommed.log 2>&1 &
/usr/bin/php -f /data/rtdamon/yunsms.class.php "$MOBILE" "recommend.py"
echo "realtime recommendation process already restarted at $t" >> $LOG_FILE
fi
#sleep `expr 3600 \* 3`
sleep `expr 60 \* 1`
done
2.3 收益
根据我们的AB测试数据来看，整体CTR提升25%。用推荐系统的A版对比无推荐的B版，用户观看时长提升47%。
3. 问题与改进
- 较多代码逻辑集中在Redis。目前Redis无灾备措施，同时IO和负载也会出现Peak。
- Spark Streaming 目前实时级别在分钟级。需要升级成storm的秒、毫秒级别。
- 需要用户点击等行为才会生产数据，容易召回不足。
