在上篇文章中我们实现了基于内容的在线召回,接下来,我们将实现基于热门文章和新文章的在线召回。主要思路是根据点击次数,统计每个频道下的热门文章,根据发布时间统计每个频道下的新文章,当推荐文章不足时,可以根据这些文章进行补足。

由于数据量较小,这里采用 Redis 存储热门文章和新文章的召回结果,数据结构如下所示

热门文章召回 结构 示例
popular_recall ch:{}:hot ch:18:hot
新文章召回 结构 示例
new_article ch:{}:new ch:18:new

热门文章存储,键为 ch:频道ID:hot 值为 分数文章ID

1
2
3
4
5
6
7
# 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:{频道ID}:new 值为 文章ID:时间戳

1
2
3
4
# 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()})

热门文章在线召回

首先,添加 Spark Streaming 和 Kafka 的配置,热门文章读取由业务系统发送到 Kafka 的 click-trace 主题中的用户实时行为数据

1
2
3
KAFKA_SERVER = "192.168.19.137:9092"
click_kafkaParams = {"metadata.broker.list": KAFKA_SERVER}
HOT_DS = KafkaUtils.createDirectStream(stream_c, ['click-trace'], click_kafkaParams)

接下来,利用 Spark Streaming 读取 Kafka 中的用户行为数据,筛选出被点击过的文章,将 Redis 中的文章热度分数进行累加即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
client = redis.StrictRedis(host=DefaultConfig.REDIS_HOST, port=DefaultConfig.REDIS_PORT, db=10)

def update_hot_redis(self):
"""
收集用户行为,更新热门文章分数
:return:
"""
def update_hot_article(rdd):
for data in rdd.collect():
# 过滤用户行为
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)

测试,写入用户行为日志

1
echo {\"actionTime\":\"2019-04-10 21:04:39\",\"readTime\":\"\",\"channelId\":18,\"param\":{\"action\": \"click\", \"userId\": \"2\", \"articleId\": \"14299\", \"algorithmCombine\": \"C2\"}} >> userClick.log

查询热门文章

1
2
3
4
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 的配置,新文章读取由业务系统发送到 Kafka 的 new-article 主题中的最新发布文章数据

1
NEW_ARTICLE_DS = KafkaUtils.createDirectStream(stream_c, ['new-article'], click_kafkaParams)

接下来,利用 Spark Streaming 读取 Kafka 的新文章,将其按频道添加到 Redis 中,Redis 的值为当前时间

1
2
3
4
5
6
7
8
9
10
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 的启动脚本中添加 new-article 主题监听配置,这样就可以收到业务系统发送过来的新文章了,重新启动 Flume 和 Kafka

1
/root/bigdata/kafka/bin/kafka-topics.sh --zookeeper 192.168.19.137:2181 --create --replication-factor 1 --topic new-article --partitions 1

测试,向 Kafka 发送新文章数据

1
2
3
4
5
6
7
8
from kafka import KafkaProducer 

# kafka消息生产者
kafka_producer = KafkaProducer(bootstrap_servers=['192.168.19.137:9092'])

# 构造消息并发送
msg = '{},{}'.format(18, 13891)
kafka_producer.send('new-article', msg.encode())

查看新文章

1
2
3
4
5
6
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,加入基于热门文章和新文章的在线召回逻辑,开启实时运行即可

1
2
3
4
5
6
7
8
9
10
11
12
13
if __name__ == '__main__':
ore = OnlineRecall()
ore.update_content_recall()
ore.update_hot_redis()
ore.update_new_redis()
stream_sc.start()
# 使用 ctrl+c 可以退出服务
_ONE_DAY_IN_SECONDS = 60 * 60 * 24
try:
while True:
time.sleep(_ONE_DAY_IN_SECONDS)
except KeyboardInterrupt:
pass

到这里,我们就完成了召回阶段的全部工作,包括基于模型和基于内容的离线召回,以及基于内容、热门文章和新文章的在线召回。通过召回,我们可以从数百万甚至上亿的原始物品数据中,筛选出和用户相关的几百、几千个可能感兴趣的物品,后面,我们将要进入到排序阶段,对召回的几百、几千个物品进行进一步的筛选和排序。

参考

https://www.bilibili.com/video/av68356229
https://pan.baidu.com/s/1-uvGJ-mEskjhtaial0Xmgw(学习资源已保存至网盘, 提取码:eakp)


【技术服务】,详情点击查看: https://mp.weixin.qq.com/s/PtX9ukKRBmazAWARprGIAg


扫一扫 关注微信公众号!号主 专注于搜索和推荐系统,尝试使用算法去更好的服务于用户,包括但不局限于机器学习,深度学习,强化学习,自然语言理解,知识图谱,还不定时分享技术,资料,思考等文章!