做网站流程 优帮云,社交网站图片展示,百度怎么对网站处罚,领优惠券的小网站怎么做在前面的文章中#xff0c;我们实现了召回和排序#xff0c;接下来将进入推荐逻辑处理阶段#xff0c;通常称为推荐中心#xff0c;推荐中心负责接收应用系统的推荐请求#xff0c;读取召回和排序的结果并进行调整#xff0c;最后返回给应用系统。推荐中心的调用流程如下…在前面的文章中我们实现了召回和排序接下来将进入推荐逻辑处理阶段通常称为推荐中心推荐中心负责接收应用系统的推荐请求读取召回和排序的结果并进行调整最后返回给应用系统。推荐中心的调用流程如下所示推荐接口设计通常推荐接口包括 Feed 流推荐和相似文章推荐Feed 流推荐根据用户偏好获取推荐文章列表(这里的时间戳用于区分是刷新推荐列表还是查看历史推荐列表)参数用户 ID频道 ID推荐文章数量请求推荐的时间戳结果曝光参数每篇文章的行为埋点参数上一条推荐的时间戳相似文章推荐当用户浏览某文章时获取该文章的相似文章列表参数文章 ID推荐文章数量结果文章 ID 列表行为埋点参数{param: {action: exposure, userId: 1, articleId: [1,2,3,4], algorithmCombine: c1},recommends: [{article_id: 1, param: {click: {action: click, userId: 1, articleId: 1, algorithmCombine: c1}, collect: ..., share: ...,read:...}},{article_id: 2, param: {click: ..., collect: ..., share: ..., read:...}},{article_id: 3, param: {click: ..., collect: ..., share: ..., read:...}},{article_id: 4, param: {click: ..., collect: ..., share: ..., read:...}}]timestamp: 1546391572}这里接口采用 gRPC 框架在 user_reco.proto 文件中定义 Protobuf 序列化协议其中定义了 Feed 流推荐接口rpc user_recommend(User) returns (Track) {} 和相似文章接口rpc article_recommend(Article) returns(Similar) {}syntax proto3;message User {string user_id 1;int32 channel_id 2;int32 article_num 3;int64 time_stamp 4;}// int32 --- int64 article_idmessage Article {int64 article_id 1;int32 article_num 2;}message param2 {string click 1;string collect 2;string share 3;string read 4;}message param1 {int64 article_id 1;param2 params 2;}message Track {string exposure 1;repeated param1 recommends 2;int64 time_stamp 3;}message Similar {repeated int64 article_id 1;}service UserRecommend {rpc user_recommend(User) returns (Track) {}rpc article_recommend(Article) returns(Similar) {}}接着通过如下命令生成服务端文件 user_reco_pb2.py 和客户端文件 user_reco_pb2_grpc.pypython -m grpc_tools.protoc -I. --python_out. --grpc_python_out. user_reco.proto定义参数解析类用于解析推荐请求的参数包括用户 ID、频道 ID、文章数量、请求时间戳以及算法名称class Temp(object):user_id -10channel_id -10article_num -10time_stamp -10algo 定义封装埋点参数方法其中参数 res 为推荐结果参数 temp 为用户请求参数将推荐结果封装为在 user_reco.proto 文件中定义的 Track 结构其中携带了文章对埋点参数包括了事件名称、算法名称以及时间等等方便后面解析用户对文章对行为信息def add_track(res, temp):封装埋点参数:param res: 推荐文章id列表:param temp: rpc参数:return: 埋点参数文章列表参数单文章参数# 添加埋点参数track {}# 准备曝光参数# 全部字符串形式提供在hive端不会解析问题_exposure {action: exposure, userId: temp.user_id, articleId: json.dumps(res),algorithmCombine: temp.algo}track[param] json.dumps(_exposure)track[recommends] []# 准备其它点击参数for _id in res:# 构造字典_dic {}_dic[article_id] _id_dic[param] {}# 准备click参数_p {action: click, userId: temp.user_id, articleId: str(_id),algorithmCombine: temp.algo}_dic[param][click] json.dumps(_p)# 准备collect参数_p[action] collect_dic[param][collect] json.dumps(_p)# 准备share参数_p[action] share_dic[param][share] json.dumps(_p)# 准备detentionTime参数_p[action] read_dic[param][read] json.dumps(_p)track[recommends].append(_dic)track[timestamp] temp.time_stampreturn trackAB Test 流量切分由于推荐算法和策略是需要不断改进和完善等所以 ABTest 也是推荐系统不可或缺的功能。可以根据用户 ID 将流量切分为多个桶(Bucket)每个桶对应一种排序策略桶内流量将使用相应的策略进行排序使用 ID 进行流量切分能够保证用户体验的一致性。通常 ABTest 过程如下所示通过定义 AB Test 参数可以实现为不同的用户使用不同的推荐算法策略其中 COMBINE 为融合方式RECALL 为召回方式SORT 为排序方式CHANNEL 为频道数量BYPASS 为分桶设置sort_dict 为不同的排序服务对象。可以看到 Algo-1 使用 LR 进行排序而 Algo-2 使用 WideDeep 进行排序from collections import namedtuple# ABTest参数信息param namedtuple(RecommendAlgorithm, [COMBINE,RECALL,SORT,CHANNEL,BYPASS])RAParam param(COMBINE{Algo-1: (1, [100, 101, 102, 103, 104], [200]), # 首页推荐所有召回结果读取LR排序Algo-2: (2, [100, 101, 102, 103, 104], [201]) # 首页推荐所有召回结果读取 排序},RECALL{100: (cb_recall, als), # 离线模型ALS召回recall:user:1115629498121 columnals:18101: (cb_recall, content), # 离线word2vec的画像内容召回 recall:user:5, content:1102: (cb_recall, online), # 在线word2vec的画像召回 recall:user:1, online:1103: new_article, # 新文章召回 redis当中 ch:18:new104: popular_article, # 基于用户协同召回结果 ch:18:hot105: (article_similar, similar) # 文章相似推荐结果 1 similar:2},SORT{200: LR,201: WDL},CHANNEL25,BYPASS[{Bucket: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, a, b, c, d],Strategy: Algo-1},{BeginBucket: [e, f],Strategy: Algo-2}])sort_dict {LR: lr_sort_service,WDL: wdl_sort_service}流量切分将用户 ID 进行哈希然后取哈希结果的第一个字符将包含该字符的策略桶所对应的算法编号赋值到此用户请求参数的 algo 属性中后面将调用该编号对应的算法策略为此用户计算推荐数据import hashlibfrom setting.default import DefaultConfig, RAParam# 进行分桶实现分流制定不同的实验策略bucket hashlib.md5(user_id.encode()).hexdigest()[:1]if bucket in RAParam.BYPASS[0][Bucket]:temp.algo RAParam.BYPASS[0][Strategy]else:temp.algo RAParam.BYPASS[1][Strategy]推荐中心逻辑推荐中心逻辑主要包括接收应用系统发送的推荐请求解析请求参数进行 ABTest 分流为用户分配推荐策略根据分配的算法调用召回服务和排序服务读取推荐结果根据业务进行调整如过滤、补足、合并信息等封装埋点参数返回推荐结果首先在 Hbase 中创建历史推荐结果表 history_recommend用于存储用户历史推荐结果create history_recommend, {NAMEchannel, TTL7776000, VERSIONS999999} 86400# 每次指定一个时间戳,可以达到不同版本的效果put history_recommend, reco:his:1, channel:18, [17283, 140357, 14668, 15182, 17999, 13648, 12884,18135]继续在 Hbase 中创建待推荐结果表 wait_recommend用于存储经过多路召回并且排序之后的待推荐结果当 wait_recommend 没有数据时才再次调用排序服务计算出新的待推荐结果并写入到 wait_recommend所以不需设置多个版本。注意该表与 cb_recall 的区别cb_recall 存储的是还未经排序的召回结果。create wait_recommend, channelput wait_recommend, reco:1, channel:18, [17283, 140357, 14668, 15182, 17999, 13648, 12884,18135]put wait_recommend, reco:1, channel:0, [17283, 140357, 14668, 15182, 17999, 13648, 12884, 17302, 13846]用户获取 Feed 流推荐数据时如果用户向下滑动发出的是刷新推荐列表的请求需要传入当前时间作为请求时间戳参数该请求时间戳必然大于 Hbase 历史推荐结果表中的请求时间戳那么程序将获取新的推荐列表并返回 Hbase 历史推荐结果表中最近一次推荐的请求时间戳用于查询历史推荐结果如果用户向上滑动发出的是查看历史推荐结果的请求需要传入前面刷新推荐列表时返回的最近一次推荐的请求时间戳该请求时间戳必然小于等于 Hbase 历史推荐结果中最近一次推荐的时间戳那么程序将获取小于等于该请求时间戳的最近一次历史推荐结果并返回小于该推荐结果最近一次推荐的时间戳也就是上一次推荐的时间戳下面是具体实现。在获取推荐列表时首先获取用户的历史数据库中最近一次时间戳 last_stamp没有则将 last_stamp 置为 0try:last_stamp self.hbu.get_table_row(history_recommend,reco:his:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode(),include_timestampTrue)[1]except Exception as e:last_stamp 0如果用户请求的时间戳小于历史推荐结果中最近一次请求的时间戳 last_stamp那么该请求为用户获取历史推荐结果1.如果没有历史推荐结果则返回时间戳 0 以及空列表 []2.如果历史推荐结果只有一条则返回这一条历史推荐结果并返回时间戳 0表示已经没有历史推荐结果(APP 可以显示已经没有历史推荐记录了)3.如果历史推荐结果有多条则返回历史推荐结果中第一条推荐结果(最近一次)然后返回历史推荐结果中第二条推荐结果的时间戳if temp.time_stamp last_stamp:try:row self.hbu.get_table_cells(history_recommend,reco:his:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode(),timestamptemp.time_stamp 1,include_timestampTrue)except Exception as e:row []res []if not row:temp.time_stamp 0res []elif len(row) 1 and row[0][1] temp.time_stamp:res eval(row[0][0])temp.time_stamp 0elif len(row) 2:res eval(row[0][0])temp.time_stamp int(row[1][1])res list(map(int, res))# 封装推荐结果track add_track(res, temp)# 曝光参数设置为空track[param] (注意这里将用户请求的时间戳 1因为 Hbase 只能获取小于该时间戳的历史推荐结果)如果用户请求的时间戳大于 Hbase 历史推荐结果中最近一次请求的时间戳 last_stamp那么该请求为用户刷新推荐列表需要读取推荐结果并返回。如果结果为空需要调用 user_reco_list() 方法再次计算推荐结果再返回。if temp.time_stamp last_stamp:# 获取缓存res redis_cache.get_reco_from_cache(temp, self.hbu)# 如果结果为空需要再次计算推荐结果 进行召回排序同时写入到hbase待推荐结果列表if not res:res self.user_reco_list(temp)temp.time_stamp int(last_stamp)track add_track(res, temp)定义 user_reco_list() 方法首先要读取多路召回结果根据为用户分配的算法策略读取相应路径的召回结果并进行重后合并reco_set []# (1, [100, 101, 102, 103, 104], [200])for number in RAParam.COMBINE[temp.algo][1]:if number 103:_res self.recall_service.read_redis_new_article(temp.channel_id)reco_set list(set(reco_set).union(set(_res)))elif number 104:_res self.recall_service.read_redis_hot_article(temp.channel_id)reco_set list(set(reco_set).union(set(_res)))else:# 100, 101, 102召回结果读取_res self.recall_service.read_hbase_recall(RAParam.RECALL[number][0],recall:user:{}.format(temp.user_id).encode(),{}:{}.format(RAParam.RECALL[number][1],temp.channel_id).encode())reco_set list(set(reco_set).union(set(_res)))接着过滤当前该请求频道的历史推荐结果如果不是 0 频道还需过滤 0 频道的历史推荐结果history_list []data self.hbu.get_table_cells(history_recommend,reco:his:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode())for _ in data:history_list list(set(history_list).union(set(eval(_))))data self.hbu.get_table_cells(history_recommend,reco:his:{}.format(temp.user_id).encode(),channel:{}.format(0).encode())for _ in data:history_list list(set(history_list).union(set(eval(_))))reco_set list(set(reco_set).difference(set(history_list)))最后根据分配的算法策略调用排序服务将分数最高的 N 个推荐结果返回并写入历史推荐结果表如果还有剩余的排序结果将其余写入待推荐结果表# 使用指定模型对召回结果进行排序# temp.user_id reco_set_sort_num RAParam.COMBINE[temp.algo][2][0]# LRreco_set sort_dict[RAParam.SORT[_sort_num]](reco_set, temp, self.hbu)if not reco_set:return reco_setelse:# 如果reco_set小于用户需要推荐的文章if len(reco_set) temp.article_num:res reco_setelse:# 大于要推荐的文章结果res reco_set[:temp.article_num]# 将剩下的文章列表写入待推荐的结果self.hbu.get_table_put(wait_recommend,reco:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode(),str(reco_set[temp.article_num:]).encode(),timestamptemp.time_stamp)# 直接写入历史记录当中表示这次又成功推荐一次self.hbu.get_table_put(history_recommend,reco:his:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode(),str(res).encode(),timestamptemp.time_stamp)return res到这里推荐中心的基本逻辑已经结束。下面是读取多路召回结果的实现细节通过指定列族读取基于模型、离线内容以及在线的召回结果并删除 cb_recall 的召回结果def read_hbase_recall_data(self, table_name, key_format, column_format):读取cb_recall当中的推荐数据读取的时候可以选择列族进行读取als, online, content:return:recall_list []data self.hbu.get_table_cells(table_name, key_format, column_format)# data是多个版本的推荐结果[[],[],[],]for _ in data:recall_list list(set(recall_list).union(set(eval(_))))self.hbu.get_table_delete(table_name, key_format, column_format)return recall_list读取 redis 中的新文章def read_redis_new_article(self, channel_id):读取新文章召回结果:param channel_id: 提供频道:return:_key ch:{}:new.format(channel_id)try:res self.client.zrevrange(_key, 0, -1)except Exception as e:res []return list(map(int, res))读取 redis 中的热门文章并选取热度最高的前 K 个文章def read_redis_hot_article(self, channel_id):读取热门文章召回结果:param channel_id: 提供频道:return:_key ch:{}:hot.format(channel_id)try:res self.client.zrevrange(_key, 0, -1)except Exception as e:# 由于每个频道的热门文章有很多因为 保留文章点击次数res list(map(int, res))if len(res) self.hot_num:res res[:self.hot_num]return res读取相似文章def read_hbase_article_similar(self, table_name, key_format, article_num):获取文章相似结果:param article_id: 文章id:param article_num: 文章数量:return:try:_dic self.hbu.get_table_row(table_name, key_format)res []_srt sorted(_dic.items(), keylambda obj: obj[1], reverseTrue)if len(_srt) article_num:_srt _srt[:article_num]for _ in _srt:res.append(int(_[0].decode().split(:)[1]))except Exception as e:res []return res使用缓存策略如果 redis 缓存中存在数据就直接从 redis 缓存中获取推荐结果如果 redis 缓存为空而 Hbase 的待推荐结果表 wait_recommend 不为空则从 wait_recommend 中获取推荐结果并将一定数量的待推荐结果放入 redis 缓存中若 redis 和 wait_recommend 都为空则需读取召回结果并进行排序将排序结果写入 Hbase 的待推荐结果表 wait_recommend 中及 redis 中(每次读取的推荐结果都要将其写入 Hbase 的历史推荐结果表 history_recommend 中)读取 redis 缓存#读取redis对应的键key reco:{}:{}:art.format(temp.user_id, temp.channel_id)# 读取删除返回结果pl cache_client.pipeline()# 读取redis数据res cache_client.zrevrange(key, 0, temp.article_num - 1)if res:# 手动删除读取出来的缓存结果pl.zrem(key, *res)如果 redis 缓存为空else:# 删除键cache_client.delete(key)try:# 从wait_recommend中读取wait_cache eval(hbu.get_table_row(wait_recommend,reco:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode()))except Exception as e:wait_cache []# 如果为空则直接返回空if not wait_cache:return wait_cache# 如果wait_recommend中有数据if len(wait_cache) 100:cache_redis wait_cache[:100]# 前100个数据放入redispl.zadd(key, dict(zip(cache_redis, range(len(cache_redis)))))# 100个后面的数据在放回wait_recommendhbu.get_table_put(wait_recommend,reco:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode(),str(wait_cache[100:]).encode())else:# 清空wait_recommend数据hbu.get_table_put(wait_recommend,reco:{}.format(temp.user_id).encode(),channel:{}.format(temp.channel_id).encode(),str([]).encode())# 所有不足100个数据放入redispl.zadd(key, dict(zip(wait_cache, range(len(wait_cache)))))res cache_client.zrange(key, 0, temp.article_num - 1)最后在 Supervisor 中配置 gRPC 实时推荐程序[program:online]environmentJAVA_HOME/root/bigdata/jdk,SPARK_HOME/root/bigdata/spark,HADOOP_HOME/root/bigdata/hadoop,PYSPARK_PYTHON/miniconda2/envs/reco_sys/bin/python ,PYSPARK_DRIVER_PYTHON/miniconda2/envs/reco_sys/bin/pythoncommand/miniconda2/envs/reco_sys/bin/python /root/toutiao_project/reco_sys/abtest/routing.pydirectory/root/toutiao_project/reco_sys/abtestuserrootautorestarttrueredirect_stderrtruestdout_logfile/root/logs/recommendsuper.logloglevelinfostopsignalKILLstopasgrouptruekillasgrouptrueThe End文章推荐系统系列到此就完结啦 撒花 若有疏漏的地方欢迎各位多多指正感谢关注love peace. 参考
