From 4230b337b402a5a76bc438b8073613b8d55538da Mon Sep 17 00:00:00 2001 From: RuyiLuo Date: Wed, 4 May 2022 13:04:07 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=89=93=E6=95=A3=E7=AD=96?= =?UTF-8?q?=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/README.md | 3 +- docs/_sidebar.md | 37 +++-- docs/ch03/ch3.2/3.2.6.md | 330 +++++++++++++++++++++++++++++++++++++++ readme.md | 3 +- 4 files changed, 350 insertions(+), 23 deletions(-) create mode 100644 docs/ch03/ch3.2/3.2.6.md diff --git a/docs/README.md b/docs/README.md index c4284923..f741a0bc 100644 --- a/docs/README.md +++ b/docs/README.md @@ -115,8 +115,7 @@ - [YoutubeDNN召回](ch03/ch3.2/3.2.4.2) - [DSSM召回](ch03/ch3.2/3.2.4.3) - [DeepFM排序模型](ch03/ch3.2/3.2.5) - - 规则与重排 - - 任务监控与调度 + - [重排(打散策略)](ch03/ch3.2/3.2.6) - **当前问题汇总** - [熟悉推荐系统基本流程问答整理](ch03/ch3.2/3.2.8.1) - [数据库的基本使用问答整理](ch03/ch3.2/3.2.8.2) diff --git a/docs/_sidebar.md b/docs/_sidebar.md index 8d634185..05bf888c 100644 --- a/docs/_sidebar.md +++ b/docs/_sidebar.md @@ -2,9 +2,9 @@ - 目录 - 第一章 推荐系统概述 - - [推荐系统的意义](/ch01/ch1.1) - - [推荐系统架构](ch01/ch1.2) - - [推荐系统技术栈](ch01/ch1.3) + - [1.1 推荐系统的意义](/ch01/ch1.1) + - [1.2 推荐系统架构](ch01/ch1.2) + - [1.3 推荐系统技术栈](ch01/ch1.3) - 第二章 推荐系统算法基础 - 2.1 经典召回模型 - 2.1.1 基于协同过滤的召回 @@ -31,7 +31,7 @@ - 2.1.5 基于树模型的召回 - [TDM](ch02/ch2.1/ch2.1.5/TDM) - 2.2 经典排序模型 - - [GBDT+LR](ch02/ch2.2/ch2.2.1) + - [2.2.1 GBDT+LR](ch02/ch2.2/ch2.2.1) - 2.2.2 特征交叉 - [FM](ch02/ch2.2/ch2.2.2/FM) - [PNN](ch02/ch2.2/ch2.2.2/PNN) @@ -55,13 +55,13 @@ - [PLE](ch02/ch2.2/ch2.2.5/PLE) - 第三章 推荐系统实战 - 3.1 天池入门赛-新闻推荐 - - [赛题理解&Baseline](ch03/ch3.1/markdown/ch3.1.1) - - [数据分析](ch03/ch3.1/markdown/ch3.1.2) - - [多路召回](ch03/ch3.1/markdown/ch3.1.3) - - [特征工程](ch03/ch3.1/markdown/ch3.1.4) - - [排序模型&模型融合](ch03/ch3.1/markdown/ch3.1.5) + - [3.1.1 赛题理解&Baseline](ch03/ch3.1/markdown/ch3.1.1) + - [3.1.2 数据分析](ch03/ch3.1/markdown/ch3.1.2) + - [3.1.3 多路召回](ch03/ch3.1/markdown/ch3.1.3) + - [3.1.4 特征工程](ch03/ch3.1/markdown/ch3.1.4) + - [3.1.5 排序模型&模型融合](ch03/ch3.1/markdown/ch3.1.5) - 3.2 新闻推荐系统的实践 - - [特别说明(必看)](ch03/ch3.2/3.2) + - [3.2.1 特别说明(必看)](ch03/ch3.2/3.2) - 3.2.1 离线物料系统的构建 - [Mysql](ch03/ch3.2/3.2.1.1) - [MongoDB](ch03/ch3.2/3.2.1.2) @@ -72,21 +72,20 @@ - [前端基础及Vue实战](ch03/ch3.2/3.2.2.1) - [flask简介及基础](ch03/ch3.2/3.2.2.2) - [前后端交互](ch03/ch3.2/3.2.2.3) - - [推荐系统流程的构建](ch03/ch3.2/3.2.3) + - [3.2.3 推荐系统流程的构建](ch03/ch3.2/3.2.3) - 3.2.4 召回 - [规则类召回](ch03/ch3.2/3.2.4.1) - [YouTubeDNN召回](ch03/ch3.2/3.2.4.2) - [DSSM召回](ch03/ch3.2/3.2.4.3) - - [DeepFM排序](ch03/ch3.2/3.2.5) - - 3.2.6 规则与重排 - - 3.2.7 任务调度与监控 + - [3.2.5 DeepFM排序](ch03/ch3.2/3.2.5) + - [3.2.6 重排(打散策略)](ch03/ch3.2/3.2.6) - 3.2.8 当前问题汇总 - [熟悉推荐系统基本流程问答整理](ch03/ch3.2/3.2.8.1) - [数据库的基本使用问答整理](ch03/ch3.2/3.2.8.2) - [离线物料系统的构建问答整理](ch03/ch3.2/3.2.8.3) - 第四章 推荐系统算法面经 - - [ML与DL基础](ch04/ch4.1) - - [推荐模型相关](ch04/ch4.2) - - [热门技术相关](ch04/ch4.3) - - [业务场景相关](ch04/ch4.4) - - [HR及其他](ch04/ch4.5) + - [4.1 ML与DL基础](ch04/ch4.1) + - [4.2 推荐模型相关](ch04/ch4.2) + - [4.3 热门技术相关](ch04/ch4.3) + - [4.4 业务场景相关](ch04/ch4.4) + - [4.5 HR及其他](ch04/ch4.5) diff --git a/docs/ch03/ch3.2/3.2.6.md b/docs/ch03/ch3.2/3.2.6.md new file mode 100644 index 00000000..4dce51dd --- /dev/null +++ b/docs/ch03/ch3.2/3.2.6.md @@ -0,0 +1,330 @@ +# 重排(打散策略） + +在精排打完分之后，一般还会有个重排阶段，这个阶段里用户的展示是最近的，所以需要考虑的问题也很多，大致分为以下三类问题： + +1. 用户体验：包括打散、多样性等 +2. 算法效率：多任务融合、实时性等 +3. 流量调控：流量扶持、生态建设等 + + + +这里主要是给出了在用户体验方面的打散策略，下面的代码实现是基于类别分桶打散的，通过不断的轮询每个类别的内容，达到内容上的打散。下面的代码打散策略还需要配合输入灌入redis中时，key的设置格式，这样便于后续打散的实现，详细的代码如下： + + + +```python +import sys +sys.path.append("../../") +sys.path.append("../") +import json +import time +import threading +from conf.dao_config import cate_dict +from conf.proj_path import bad_case_news_log_path +from dao.redis_server import RedisServer +from dao.mysql_server import MysqlServer +from dao.entity.register_user import RegisterUser +from controller.user_action_controller import UserAction +from collections import defaultdict + +redis_server = RedisServer() + +class OnlineServer(object): + """单例模式推荐服务类 + """ + _instance_lock = threading.Lock() + + def __init__(self,): + self.reclist_redis_db = redis_server.get_reclist_redis() + self.static_news_info_redis_db = redis_server.get_static_news_info_redis() + self.dynamic_news_info_redis_db = redis_server.get_dynamic_news_info_redis() + self.exposure_redis_db = redis_server.get_exposure_redis() + self.register_sql_sess = MysqlServer().get_register_user_session() + self.cate_dict = cate_dict + self.cate_id_list = list(self.cate_dict.keys()) + self.bad_case_news_log_path = bad_case_news_log_path + self.name2id_cate_dict = {v: k for k, v in self.cate_dict.items()} + self._set_user_group() + + def __new__(cls, *args, **kwargs): + if not hasattr(OnlineServer, "_instance"): + with OnlineServer._instance_lock: + if not hasattr(OnlineServer, "_instance"): + OnlineServer._instance = object.__new__(cls) + return OnlineServer._instance + + def _get_register_user_cold_start_redis_key(self, userid): + """通过查sql表得到用户的redis key进而确定当前新用户使用哪一个新的模板 + """ + user_info = self.register_sql_sess.query(RegisterUser).filter(RegisterUser.userid == userid).first() + print(user_info) + if int(user_info.age) < 23 and user_info.gender == "female": + redis_key = "cold_start_group:{}".format(str(1)) + elif int(user_info.age) >= 23 and user_info.gender == "female": + redis_key = "cold_start_group:{}".format(str(2)) + elif int(user_info.age) < 23 and user_info.gender == "male": + redis_key = "cold_start_group:{}".format(str(3)) + elif int(user_info.age) >= 23 and user_info.gender == "male": + redis_key = "cold_start_group:{}".format(str(4)) + else: + pass + return redis_key + + def _set_user_group(self): + """将用户进行分组 + 1. age < 23 && gender == female + 2. age >= 23 && gender == female + 3. age < 23 && gender == male + 4. age >= 23 && gender == male + """ + self.user_group = { + "1": ["国内","娱乐","体育","科技"], + "2": ["国内","社会","美股","财经","股市"], + "3": ["国内","股市","体育","科技"], + "4": ["国际", "国内","军事","社会","美股","财经","股市"] + } + self.group_to_cate_id_dict = defaultdict(list) + for k, cate_list in self.user_group.items(): + for cate in cate_list: + self.group_to_cate_id_dict[k].append(self.name2id_cate_dict[cate]) + + def _get_register_user_group_id(self, age, gender): + """获取注册用户的分组, + bug: 新用户注册可能会有延迟 + """ + if int(age) < 23 and gender == "female": + return "1" + elif int(age) >= 23 and gender == "female": + return "2" + elif int(age) < 23 and gender == "male": + return "3" + elif int(age) >= 23 and gender == "male": + return "4" + else: + return "error" + + def _copy_cold_start_list_to_redis(self, user_id, group_id): + """将确定分组后的用户的物料添加到redis中，并记录当前用户的所有新闻类别id + """ + # 遍历当前分组的新闻类别 + for cate_id in self.group_to_cate_id_dict[group_id]: + group_redis_key = "cold_start_group:{}:{}".format(group_id, cate_id) + user_redis_key = "cold_start_user:{}:{}".format(user_id, cate_id) + self.reclist_redis_db.zunionstore(user_redis_key, [group_redis_key]) + # 将用户的类别集合添加到redis中 + cate_id_set_redis_key = "cold_start_user_cate_set:{}".format(user_id) + self.reclist_redis_db.sadd(cate_id_set_redis_key, *self.group_to_cate_id_dict[group_id]) + + def _judge_and_get_user_reverse_index(self, user_id, rec_type, age=None, gender=None): + """判断当前用户是否存在倒排索引, 如果没有的话拷贝一份 + """ + if rec_type == 'hot_list': + # 判断用户是否存在热门列表 + cate_id = self.cate_id_list[0] # 随机选择一个就行 + hot_list_user_key = "user_id_hot_list:{}:{}".format(str(user_id), cate_id) + if self.reclist_redis_db.exists(hot_list_user_key) == 0: + # 给用户拷贝一份每个类别的倒排索引 + for cate_id in self.cate_id_list: + cate_id_news_templete_key = "hot_list_news_cate:{}".format(cate_id) + hot_list_user_key = "user_id_hot_list:{}:{}".format(str(user_id), cate_id) + self.reclist_redis_db.zunionstore(hot_list_user_key, [cate_id_news_templete_key]) + elif rec_type == "cold_start": + # 判断用户是否在冷启动列表中 + cate_id_set_redis_key = "cold_start_user_cate_set:{}".format(user_id) + print("判断用户是否在冷启动列表中 {}".format(self.reclist_redis_db.exists(cate_id_set_redis_key))) + if self.reclist_redis_db.exists(cate_id_set_redis_key) == 0: + # 如果系统中没有当前用户的冷启动倒排索引, 那么就需要从冷启动模板中复制一份 + # 确定用户分组 + try: + group_id = self._get_register_user_group_id(age, gender) + except: + return False + print("group_id : {}".format(group_id)) + self._copy_cold_start_list_to_redis(user_id, group_id) + else: + pass + return True + + def _get_user_expose_set(self, user_id): + """获取用户曝光列表 + """ + user_exposure_prefix = "user_exposure:" + user_exposure_key = user_exposure_prefix + str(user_id) + # 获取用户当前曝光列表 + if self.exposure_redis_db.exists(user_exposure_key) > 0: + exposure_list = self.exposure_redis_db.smembers(user_exposure_key) + news_expose_set = set(map(lambda x: x.split(':')[0], exposure_list)) + else: + news_expose_set = set() + return news_expose_set + + def _save_user_exposure(self, user_id, newslist): + """记录用户曝光到redis""" + if len(newslist) == 0: return False # 无曝光数目 + + ctime = str(round(time.time()*1000)) # 曝光时间戳 + key = "user_exposure:" + str(user_id) # 为key拼接 + # 将历史曝光记录与newlist(最新曝光)的交集新闻提出来 并将该部分删除，防止重复存储曝光新闻 + exposure_news_set = self.exposure_redis_db.smembers(key) # 历史曝光记录 + + del_exposure_news = [] # 历史曝光记录与newlist(最新曝光)的交集新闻,需要删除 + if exposure_news_set.__len__() != 0: + del_exposure_news = [item for item in exposure_news_set if item.split(":")[0] in newslist] + + # 为曝光过的新闻拼接时间 + news_save = [] + for news_id in newslist: + val = news_id+":"+ctime + val = val.replace('"', "'" ) # 将双引号都替换成单引号 + news_save.append(val) + + # 存储redis + try: + if del_exposure_news.__len__() != 0: + self.exposure_redis_db.srem(key,*del_exposure_news) + self.exposure_redis_db.sadd(key,*news_save) + except Exception as e: + print(str(e)) + return False + return True + + def _get_polling_rec_list(self, user_id, news_expose_set, cate_id_list, rec_type, one_page_news_cnt=10): + """获取轮询的打散新闻列表 + """ + # 候选曝光列表 + exposure_news_list = [] + # 用户展示新闻列表 + user_news_list = [] + iter_cnt = 0 + # 给每个用户轮询每个类别的新闻，获取打散之后的新闻列表 + while len(user_news_list) != one_page_news_cnt: + cate_id_index = iter_cnt % len(cate_id_list) + cate_id = cate_id_list[cate_id_index] + if rec_type == "hot_list": + user_redis_key = "user_id_hot_list:{}:{}".format(str(user_id), cate_id) + elif rec_type == "cold_start": + user_redis_key = "cold_start_user:{}:{}".format(str(user_id), cate_id) + else: + pass + cur_cate_cnt = 0 + while self.reclist_redis_db.zcard(user_redis_key) > 0: + # 摘取排名第一的新闻 + news_id_and_cate = self.reclist_redis_db.zrevrange(user_redis_key, 0, 0)[0] + news_id = news_id_and_cate.split('_')[1] # 将新闻id切分出来 + if news_id in news_expose_set: + # 将当前新闻id添加到待删除的新闻列表中 + self.reclist_redis_db.zrem(user_redis_key, news_id_and_cate) + continue + # TODO 在数据入库的时候离线处理无法成功加载json的问题 + # 获取新闻详细信息, 由于爬取的新闻没有做清理，导致有些新闻无法转化成json的形式 + # 所以这里如果转化失败的内容也直接删除掉 + try: + news_info_dict = self.get_news_detail(news_id) + cur_cate_cnt += 1 + except Exception as e: + # 删除无效的新闻 + self.reclist_redis_db.zrem(user_redis_key, news_id_and_cate) + # 记录无效的新闻的id + with open(self.bad_case_news_log_path, "a+") as f: + f.write(news_id + "\n") + print("there are not news detail info for {}".format(news_id)) + continue + # 删除当前key + self.reclist_redis_db.zrem(user_redis_key, news_id_and_cate) + # 判断当前类别的新闻是否摘取成功, 摘取成功的话就推出当前循环 + if cur_cate_cnt == 1: + # 将摘取成功的新闻信息添加到用户新闻列表中 + user_news_list.append(news_info_dict) + exposure_news_list.append(news_id) + break + iter_cnt += 1 + return user_news_list, exposure_news_list + + def get_cold_start_rec_list_v2(self, user_id, age=None, gender=None): + """推荐页展示列表，使用轮询的方式进行打散 + """ + # 获取用户曝光列表 + news_expose_set = self._get_user_expose_set(user_id) + + # 判断用户是否存在冷启动列表中 + flag = self._judge_and_get_user_reverse_index(user_id, "cold_start", age, gender) + + if not flag: + print("_judge_and_get_user_reverse_index fail") + return [] + + # 获取用户的cate id列表 + cate_id_set_redis_key = "cold_start_user_cate_set:{}".format(user_id) + cate_id_list = list(self.reclist_redis_db.smembers(cate_id_set_redis_key)) + + # 通过轮询的方式 + user_news_list, exposure_news_list = self._get_polling_rec_list(user_id, news_expose_set, cate_id_list, rec_type="cold_start") + + # 添加曝光内容 + self._save_user_exposure(user_id, exposure_news_list) + return user_news_list + + def get_hot_list_v2(self, user_id): + """热门页展示列表，使用轮询的方式进行打散 + """ + # 获取用户曝光列表 + news_expose_set = self._get_user_expose_set(user_id) + + # 判断用户是否存在热门列表 + self._judge_and_get_user_reverse_index(user_id, "hot_list") + + # 通过轮询的方式获取用户的展示列表 + user_news_list, exposure_news_list = self._get_polling_rec_list(user_id, news_expose_set, self.cate_id_list, rec_type="hot_list") + + # 添加曝光内容 + self._save_user_exposure(user_id, exposure_news_list) + return user_news_list + + def get_news_detail(self, news_id): + """获取新闻展示的详细信息 + """ + news_info_str = self.static_news_info_redis_db.get("static_news_detail:" + news_id) + news_info_str = news_info_str.replace('\'', '\"' ) # 将单引号都替换成双引号 + news_info_dit = json.loads(news_info_str) + news_dynamic_info_str = self.dynamic_news_info_redis_db.get("dynamic_news_detail:" + news_id) + news_dynamic_info_str = news_dynamic_info_str.replace("'", '"' ) # 将单引号都替换成双引号 + news_dynamic_info_dit = json.loads(news_dynamic_info_str) + for k in news_dynamic_info_dit.keys(): + news_info_dit[k] = news_dynamic_info_dit[k] + return news_info_dit + + def update_news_dynamic_info(self, news_id,action_type): + """更新新闻展示的详细信息 + """ + news_dynamic_info_str = self.dynamic_news_info_redis_db.get("dynamic_news_detail:" + news_id) + news_dynamic_info_str = news_dynamic_info_str.replace("'", '"' ) # 将单引号都替换成双引号 + news_dynamic_info_dict = json.loads(news_dynamic_info_str) + if len(action_type) == 2: + if action_type[1] == "true": + news_dynamic_info_dict[action_type[0]] +=1 + elif action_type[1] == "false": + news_dynamic_info_dict[action_type[0]] -=1 + else: + news_dynamic_info_dict["read_num"] +=1 + news_dynamic_info_str = json.dumps(news_dynamic_info_dict) + news_dynamic_info_str = news_dynamic_info_str.replace('"', "'" ) + res = self.dynamic_news_info_redis_db.set("dynamic_news_detail:" + news_id, news_dynamic_info_str) + return res + + def test(self): + user_info = self.register_sql_sess.query(RegisterUser).filter(RegisterUser.userid == "4566566568405766145").first() + print(user_info.age) + + +if __name__ == "__main__": + # 测试单例模式 + oneline_server = OnlineServer() + # oneline_server.get_hot_list("4563333734895456257") + oneline_server.test() +``` + + + +**参考资料** + +[推荐算法架构4：重排](https://xieyangyi.blog.csdn.net/article/details/123095982) \ No newline at end of file diff --git a/readme.md b/readme.md index 5074e7fb..74260d66 100644 --- a/readme.md +++ b/readme.md @@ -117,8 +117,7 @@ - [YoutubeDNN召回](docs/ch03/ch3.2/3.2.4.2.md) - [DSSM召回](docs/ch03/ch3.2/3.2.4.3.md) - [DeepFM排序模型](docs/ch03/ch3.2/3.2.5.md) - - 规则与重排 - - 任务监控与调度 + - [重排(打散策略)](docs/ch03/ch3.2/3.2.6.md) - **当前问题汇总** - [熟悉推荐系统基本流程问答整理](docs/ch03/ch3.2/3.2.8.1.md) - [数据库的基本使用问答整理](docs/ch03/ch3.2/3.2.8.2.md)