推薦閱讀：
文章推薦系統(tǒng) | 一鼻弧、推薦流程設(shè)計(jì)
文章推薦系統(tǒng) | 二、同步業(yè)務(wù)數(shù)據(jù)
文章推薦系統(tǒng) | 三惯裕、收集用戶行為數(shù)據(jù)
文章推薦系統(tǒng) | 四温数、構(gòu)建離線文章畫(huà)像
文章推薦系統(tǒng) | 五、計(jì)算文章相似度
文章推薦系統(tǒng) | 六蜻势、構(gòu)建離線用戶畫(huà)像
文章推薦系統(tǒng) | 七撑刺、構(gòu)建離線文章特征和用戶特征
文章推薦系統(tǒng) | 八、基于模型的離線召回
文章推薦系統(tǒng) | 九握玛、基于內(nèi)容的離線及在線召回
在上篇文章中我們實(shí)現(xiàn)了基于內(nèi)容的在線召回够傍，接下來(lái)甫菠，我們將實(shí)現(xiàn)基于熱門(mén)文章和新文章的在線召回。主要思路是根據(jù)點(diǎn)擊次數(shù)冕屯，統(tǒng)計(jì)每個(gè)頻道下的熱門(mén)文章寂诱，根據(jù)發(fā)布時(shí)間統(tǒng)計(jì)每個(gè)頻道下的新文章，當(dāng)推薦文章不足時(shí)安聘，可以根據(jù)這些文章進(jìn)行補(bǔ)足痰洒。
由于數(shù)據(jù)量較小，這里采用 Redis 存儲(chǔ)熱門(mén)文章和新文章的召回結(jié)果浴韭，數(shù)據(jù)結(jié)構(gòu)如下所示
| 熱門(mén)文章召回 | 結(jié)構(gòu) | 示例 |
|---|---|---|
| popular_recall | ch:{}:hot | ch:18:hot |
| 新文章召回 | 結(jié)構(gòu) | 示例 |
|---|---|---|
| new_article | ch:{}:new | ch:18:new |
熱門(mén)文章存儲(chǔ)丘喻，鍵為 ch:頻道ID:hot 值為 分?jǐn)?shù) 和 文章ID
# ZINCRBY key increment member
# ZSCORE
# 為有序集 key 的成員 member 的 score 值加上增量 increment 。
client.zincrby("ch:{}:hot".format(row['channelId']), 1, row['param']['articleId'])
# ZREVRANGE key start stop [WITHSCORES]
client.zrevrange(ch:{}:new, 0, -1)
新文章存儲(chǔ)念颈，鍵為 ch:{頻道ID}:new 值為 文章ID:時(shí)間戳
# ZADD ZRANGE
# ZADD key score member [[score member] [score member] ...]
# ZRANGE page_rank 0 -1
client.zadd("ch:{}:new".format(channel_id), {article_id: time.time()})
熱門(mén)文章在線召回
首先泉粉，添加 Spark Streaming 和 Kafka 的配置，熱門(mén)文章讀取由業(yè)務(wù)系統(tǒng)發(fā)送到 Kafka 的 click-trace 主題中的用戶實(shí)時(shí)行為數(shù)據(jù)
KAFKA_SERVER = "192.168.19.137:9092"
click_kafkaParams = {"metadata.broker.list": KAFKA_SERVER}
HOT_DS = KafkaUtils.createDirectStream(stream_c, ['click-trace'], click_kafkaParams)
接下來(lái)榴芳，利用 Spark Streaming 讀取 Kafka 中的用戶行為數(shù)據(jù)嗡靡，篩選出被點(diǎn)擊過(guò)的文章，將 Redis 中的文章熱度分?jǐn)?shù)進(jìn)行累加即可
client = redis.StrictRedis(host=DefaultConfig.REDIS_HOST, port=DefaultConfig.REDIS_PORT, db=10)
def update_hot_redis(self):
"""
收集用戶行為窟感，更新熱門(mén)文章分?jǐn)?shù)
:return:
"""
def update_hot_article(rdd):
for data in rdd.collect():
# 過(guò)濾用戶行為
if data['param']['action'] in ['exposure', 'read']:
pass
else:
client.zincrby("ch:{}:hot".format(data['channelId']), 1, data['param']['articleId'])
HOT_DS.map(lambda x: json.loads(x[1])).foreachRDD(update_hot_article)
測(cè)試讨彼，寫(xiě)入用戶行為日志
echo {\"actionTime\":\"2019-04-10 21:04:39\",\"readTime\":\"\",\"channelId\":18,\"param\":{\"action\": \"click\", \"userId\": \"2\", \"articleId\": \"14299\", \"algorithmCombine\": \"C2\"}} >> userClick.log
查詢熱門(mén)文章
127.0.0.1:6379[10]> keys *
1) "ch:18:hot"
127.0.0.1:6379[10]> ZRANGE "ch:18:hot" 0 -1
1) "14299"
新文章在線召回
首先，添加 Spark Streaming 和 Kafka 的配置肌括，新文章讀取由業(yè)務(wù)系統(tǒng)發(fā)送到 Kafka 的 new-article 主題中的最新發(fā)布文章數(shù)據(jù)
NEW_ARTICLE_DS = KafkaUtils.createDirectStream(stream_c, ['new-article'], click_kafkaParams)
接下來(lái)点骑，利用 Spark Streaming 讀取 Kafka 的新文章酣难，將其按頻道添加到 Redis 中谍夭，Redis 的值為當(dāng)前時(shí)間
def update_new_redis(self):
"""更新頻道最新文章
:return:
"""
def add_new_article(rdd):
for row in rdd.collect():
channel_id, article_id = row.split(',')
client.zadd("ch:{}:new".format(channel_id), {article_id: time.time()})
NEW_ARTICLE_DS.map(lambda x: x[1]).foreachRDD(add_new_article)
還需要在 Kafka 的啟動(dòng)腳本中添加 new-article 主題監(jiān)聽(tīng)配置，這樣就可以收到業(yè)務(wù)系統(tǒng)發(fā)送過(guò)來(lái)的新文章了憨募，重新啟動(dòng) Flume 和 Kafka
/root/bigdata/kafka/bin/kafka-topics.sh --zookeeper 192.168.19.137:2181 --create --replication-factor 1 --topic new-article --partitions 1
測(cè)試紧索，向 Kafka 發(fā)送新文章數(shù)據(jù)
from kafka import KafkaProducer
# kafka消息生產(chǎn)者
kafka_producer = KafkaProducer(bootstrap_servers=['192.168.19.137:9092'])
# 構(gòu)造消息并發(fā)送
msg = '{},{}'.format(18, 13891)
kafka_producer.send('new-article', msg.encode())
查看新文章
127.0.0.1:6379[10]> keys *
1) "ch:18:hot"
2) "ch:18:new"
127.0.0.1:6379[10]> ZRANGE "ch:18:new" 0 -1
1) "13890"
2) "13891"
最后，修改 online_update.py菜谣，加入基于熱門(mén)文章和新文章的在線召回邏輯珠漂，開(kāi)啟實(shí)時(shí)運(yùn)行即可
if __name__ == '__main__':
ore = OnlineRecall()
ore.update_content_recall()
ore.update_hot_redis()
ore.update_new_redis()
stream_sc.start()
# 使用 ctrl+c 可以退出服務(wù)
_ONE_DAY_IN_SECONDS = 60 * 60 * 24
try:
while True:
time.sleep(_ONE_DAY_IN_SECONDS)
except KeyboardInterrupt:
pass
到這里，我們就完成了召回階段的全部工作尾膊，包括基于模型和基于內(nèi)容的離線召回媳危，以及基于內(nèi)容、熱門(mén)文章和新文章的在線召回冈敛。通過(guò)召回待笑，我們可以從數(shù)百萬(wàn)甚至上億的原始物品數(shù)據(jù)中，篩選出和用戶相關(guān)的幾百抓谴、幾千個(gè)可能感興趣的物品暮蹂，后面寞缝，我們將要進(jìn)入到排序階段，對(duì)召回的幾百仰泻、幾千個(gè)物品進(jìn)行進(jìn)一步的篩選和排序荆陆。
參考
https://www.bilibili.com/video/av68356229
https://pan.baidu.com/s/1-uvGJ-mEskjhtaial0Xmgw（學(xué)習(xí)資源已保存至網(wǎng)盤(pán)， 提取碼：eakp）
