✨写在前面：强烈推荐给大家一个优秀的人工智能学习网站，内容包括人工智能基础、机器学习、深度学习神经网络等，详细介绍各部分概念及实战教程，通俗易懂，非常适合人工智能领域初学者及研究者学习。➡️点击跳转到网站。

---

➡️前一篇新闻推荐系统系列参考：

https://blog.csdn.net/qq_36816848/article/details/121941803

📝音乐推荐项目描述：

音乐推荐系统demo，针对用户画像、用户行为及物品特征构建推荐系统。使用cb、cf算法做推荐召回，使用redis数据库做缓存处理，结合LR算法推荐排序，达到粗排、精排效果，实现个性化推荐。

🌐音乐推荐系统项目链接:

https://github.com/GoAlers/music-top-recommend

一、推荐项目流程：

1.预处理阶段：

对三个数据进行预处理，合并用户与物品相关信息，处理后的merge_base.data数据字段包含itemid、userid、用户信息(年龄、性别、收入、地区)、物品信息（名字、描述、时长、标签）、用户行为数据(收听时长)等。

2.粗排召回阶段

使用CB算法，基于内容进行jieba中文分词，计算itemid对应分词的tfidf分数，整理训练数据；使用mr协同过滤进行相关性计算，训练得到物品之间对应分数item-item能得到II矩阵；而CF算法则通过协同过滤将UI矩阵转成II矩阵。最后，将格式化后数据结果按k/v形式批量灌入redis数据库。

3.精排阶段

利用LR算法进行推荐排序，得到权重w、b用于模型构建rank_model。结合用户与物品标签获取用户与物品特征训练数据。

4.推荐流程阶段

加载特征数据及排序模型，检索redis数据库获取候选集，利用逻辑回归sigmoid函数打分并排序，最终利用可视化页面实现itemid->name进行top10评分相关推荐。

二、推荐系统思路框架

流程思路：

1、数据预处理（用户画像数据、物品元数据、用户行为数据）
2、召回（CB、CF算法）
3、LR训练模型的数据准备，即用户特征数据，物品特征数据
4、模型准备，即通过LR算法训练模型数据得到w，b
5、推荐系统流程：初始化加载data，获取user_feature.data，item_feature.data特征 ，用字典维护。

（1）解析请求：userid，itemid
（2）加载模型：加载排序模型（model.w，model.b）
（3）检索候选集合：利用cb，cf去redis里面检索数据库，得到候选集合
（4）获取用户特征user_feature.data ：userid
（5）获取物品特征item_feature.data ：itemid
（6）打分(逻辑回归函数sigmoid: 1 / (1 + exp(-wx)))，排序
（7）top-n过滤（精排）
（8）数据包装（itemid->name），返回

数据集： 默认以’\001’分割

1.用户画像数据（user_profile.data）
字段：userid, gender, age, salary, location

userid，性别，年龄，收入，地域
00dee4bd83b6c115b865a64e92745000,男,26-35,10000-20000,台湾
000798c9a4cab6d6b3065e287d917000,女,36-45,5000-10000,陕西
00a2fe08da2621d2e9dd3ff7f89e7000,女,19-25,20000-100000,宁夏
009cb9b50e1c38e6e8d70b9accc08000,男,26-35,10000-20000,新疆
0182020bfb3e321a852abc857e2c9000,男,46-100,10000-20000,山西

2.物品元数据（music_meta）
字段：itemid, name, desc, total_timelen, location, tags

itemid,name,内容 ,时长，地域，标签
1084709100 outta my head 周杰伦MV-范特西 227  周杰伦MV
1119509100 GEE 少女时代 舞蹈，舞蹈教学 197  舞蹈，舞蹈教学,流行
1126809100 SPICA - You Don't Love Me 敬请期待先上预告 24  他不爱我预告
1144209100TAE MIN  -  Steps 总理和我在OST 218  东方神起TAE,流行

3.用户行为数据（user_watch_pref.sml）
userid，itemid，用户对item收听时长，点击时间（小时）

01e3fdf415107cd6046a07481fbed499 6470209102 555 16
01e3fdf415107cd6046a07481fbed499 6470209102 2024 22
012e5c128fbe16c302c9a12f9238f871 6470209102 1047 13
012e5c128fbe16c302c9a12f9238f871 6470209102 1155 10
01e3fdf415107cd6046a07481fbed499 6470209102 795 22

三、推荐系统步骤

1、数据预处理：

总体思路：处理原始的数据，将用户画像数据 、物品元数据、用户行为数据，3份融合到一起，得到处理后merge_base.data，用于cb、cf算法进行计算。

python gen_base.py

#coding=utf-8import sys#找到三类原始数据文件，用户画像数据、物品元数据，用户行为数据
user_action_data = '../data/user_watch_pref.sml'
music_meta_data = '../data/music_meta'
user_profile_data = '../data/user_profile.data'##将合并后的元数据放到新的文件里
output_file = '../data/merge_base.data'# 将3份数据merge后的结果输出，供下游数据处理
ofile = open(output_file, 'w')# step 1. 处理物品元数据，将处理后的结果放入字典里面，key是itemid，value为物品对应的信息，为最后写入做准备
item_info_dict = {}
with open(music_meta_data, 'r') as fd:for line in fd:ss = line.strip().split('\001')if len(ss) != 6:continueitemid, name, desc, total_timelen, location, tags = ssitem_info_dict[itemid] = '\001'.join([name, desc, total_timelen, location, tags])# step 2. 处理用户画像数据，将处理后的结果放入字典里面，key是用户id，value是用户信息
user_profile_dict = {}
with open(user_profile_data, 'r') as fd:for line in fd:ss = line.strip().split(',')if len(ss) != 5:continueuserid, gender, age, salary, location = ssuser_profile_dict[userid] = '\001'.join([gender, age, salary, location])# step 3. 写入最后的信息，将用户行为数据进行处理，把step1和step2得到的数据一并归纳在文件里面
with open(user_action_data, 'r') as fd:for line in fd:ss = line.strip().split('\001')if len(ss) != 4:continueuserid, itemid, watch_len, hour = ssif userid not in user_profile_dict:continueif itemid not in item_info_dict:continueofile.write('\001'.join([userid, itemid, watch_len, hour, \user_profile_dict[userid], item_info_dict[itemid]]))ofile.write("\n")ofile.close()

合并后得到类似下面数据merge_base.data
数据字段包含itemid、userid、用户信息(年龄、性别、收入、地区)、物品信息（名字、描述、时长、标签）、用户行为数据(收听时长)

2、【召回】CB算法

【注：CF、CB只是数据形态不一样，逻辑是一样的，都是计算矩阵两两元素之间的分数，即相关性,都可以套用协同过滤模板得到ii矩阵】
进一步说明：最后得到基于cb的 II 矩阵，输出得到 cb_result 数据，即 mr_cf 程序（只是数据源变了）。在这里输入数据是 cb_train.data ，内容是 token、itemid、score ，相当于CB算法基于item属性的CB，输出则是itemid、itemid、score。
即：正排表：item ——> token、token、token
倒排表：token ——> item、item、item
CF:user,item，score -》item item sim
CB:token，item，score -》item item sim

总体思路：将初始化好的用户，物品，用户行为数据进行处理，对item去重，目的是为了得到token，itemid，score，对于生成的数据里面的name，将itemName进行分词，得到tfidf权重，同时将desc进行分词，处理name和desc。

代码思路：元数据中tags已经分类好无需再次进行切分，只需要用idf词表查处权重即可，针对name、desc、tags三个分词结果，适当调整name权重比例，分别对这三类得出的分数再次进行分数权重划分，最后得到cb的初始数据。

（1）利用jieba分词，对item name进行中文分词，item —> name desc tag ，对每个item的这三个字段进行分词并计算score，将数据用字典token_dict维护执行python gen_cb_train.py

#coding=utf-8import sys
sys.path.append('../')
reload(sys)
sys.setdefaultencoding('utf-8')import jieba
import jieba.posseg
import jieba.analyse#读入初始数据
input_file = "../data/merge_base.data"# 输出cb训练数据
output_file = '../data/cb_train.data'
outfile = open(output_file, 'w')#定义三类的权重分数（大小可自行设定）
RATIO_FOR_NAME = 0.9
RATIO_FOR_DESC = 0.1
RATIO_FOR_TAGS = 0.05#为tags读入idf权重值
idf_file = '../data/idf.txt'
idf_dict = {}
with open(idf_file, 'r') as fd:for line in fd:token, idf_score = line.strip().split(' ')idf_dict[token] = idf_score#开始处理初始数据
itemid_set = set()
with open(input_file, 'r') as fd:for line in fd:ss = line.strip().split('\001')# 用户行为userid = ss[0].strip()itemid = ss[1].strip()watch_len = ss[2].strip()hour = ss[3].strip()# 用户画像gender = ss[4].strip()age = ss[5].strip()salary = ss[6].strip()user_location = ss[7].strip()# 物品元数据name = ss[8].strip()desc = ss[9].strip()total_timelen = ss[10].strip()item_location = ss[11].strip()tags = ss[12].strip()# 对item去重，相同的itemid不用再计算，因为都一样，这里用到continue特性，当不同的时候才继续执行下面的代码if itemid not in itemid_set:itemid_set.add(itemid)else:continue# 去掉重复后的itemid，然后我们进行分词，计算权重，放到字典里面token_dict = {}#对name统计for a in jieba.analyse.extract_tags(name, withWeight=True):token = a[0]score = float(a[1])token_dict[token] = score * RATIO_FOR_NAME#对desc进行分词，这里需要注意的是描述一般会含有name中的词，这里我们把有的词的分数进行相加，没有的放入for a in jieba.analyse.extract_tags(desc, withWeight=True):token = a[0]score = float(a[1])if token in token_dict:token_dict[token] += score * RATIO_FOR_DESCelse:token_dict[token] = score * RATIO_FOR_DESC# 对tags 进行分数计算for tag in tags.strip().split(','):if tag not in idf_dict:continueelse:if tag in token_dict:token_dict[tag] += float(idf_dict[tag]) * RATIO_FOR_TAGSelse:token_dict[tag] = float(idf_dict[tag]) * RATIO_FOR_TAGS#循环遍历token_dict，输出toke，itemid，scorefor k, v in token_dict.items():token = k.strip()score = str(v)ofile.write(','.join([token, itemid, score]))ofile.write("\n")outfile.close()

经过数据预处理，得到如下格式的cb训练数据：
tokenid itemid，score（itemid中的各个token在该item中的重要性）
哲,4090309101,0.896607562717
大连,4090309101,0.568628215367
舞曲,4090309101,0.713898826298
大美妞,4090309101,0.896607562717
网络,4090309101,0.465710816584
伤感,4090309101,0.628141853463
（注：最后一个字段不是传统的TF-IDF,因为分出的词在name,desc,tag里面权重不同，即切分单词在desc中重要性不同)
　　
（2）用协同过滤算法跑出item-item数据
　　
求相似度的II矩阵（相似的item配对，形成II矩阵）。

相似度计算：要用到MapReduce的框架来进行，只要用到shuffle阶段，对map出来的结果排序，reduce进行两两配对，主要是wordcount逻辑，主要说下注意的部分：我们需要把两两分数的过滤掉，或是把itemA和itemB相同的item过滤掉，因为这部分数据没有任何意义。

1.map阶段：

总体思路：这里需要把初始化后的结果进行map排序，为了后续两两取 pair对，所以这里我们需要进行map，其实什么也不用操作输出即可

import sys
import refor line in sys.stdin:ss = line.strip().split(',')if len(ss) != 3:continuer1 = u'[a-zA-Z0-9’!"#$%&\'()*+,-./:;<=>?@，。?★、…【】《》？“”‘’！[\\]^_`{|}~]+'ss[0] = re.sub(r1,'',ss[0])if len(ss[0]) == 0:continueprint ','.join([ss[0], ss[1], ss[2]])

2.reduce阶段：

在pair reduce之前做过map操作，以token，item，score输出，所以token排序好的， 这里我们相当于求的是II矩阵，所以是相同的token的item进行相似度计算

总体思路：1、进行user统计，若相同，把相同的user的item和score放入list里面2、不相同，开始进行两两配对，循环该list，进行两两配对，求出相似度

import  sys
import  mathcur_token = None
item_score_list = []
for line in sys.stdin:ss = line.strip().split(',')itemid = ss[1]score = float(ss[2])if len(ss) != 3:continueif cur_token == None:cur_token = ss[0]if cur_token != ss[0]:#这里需要注意的是range的区间前闭后开，同时注意range中即使前闭后开，刚开始是从0即列表里面的第一个，循环到列表最后一个的前一个for i in range(0,len(item_score_list)-1):for j in range(i+1,len(item_score_list)):item_a,score_a = item_score_list[i]item_b,score_b = item_score_list[j]#score = float(score_a * score_b)/float(math.sqrt(pow(score_a,2))*math.sqrt(pow(score_b,2)))#输出两遍的目的是为了形成II矩阵的对称score = float(score_a*score_b)if item_a == item_b:continueif score < 0.08:continueprint "%s\t%s\t%s" % (item_a, item_b, score)print "%s\t%s\t%s" % (item_b, item_a, score)cur_token = ss[0]item_score_list = []item_score_list.append((itemid,float(score)))for i in range(0, len(item_score_list) - 1):for j in range(i + 1, len(item_score_list)):item_a, score_a = item_score_list[i]item_b, score_b = item_score_list[j]#score = (score_a * score_b) / (math.sqrt(pow(score_a, 2)) * math.sqrt(pow(score_b, 2))# 输出两遍的目的是为了形成II矩阵的对称score = float(score_a * score_b)if item_a == item_b:continueif score < 0.08:continueprint "%s\t%s\t%s" % (item_a, item_b, score)print "%s\t%s\t%s" % (item_b, item_a, score)

最后得到基于cb的ii矩阵

（3）对数据格式化，item-> item list形式，整理出KV形式
　　python gen_reclist.py

思路：上步通过CB算法得到itemA，itemB，score需要把放入到redis库，先进行数据格式化，以itemA为key与itemA有相似度的itemB，和分数，以value的形式存入内存库
1、创建一个字典，将key放入itemA，value 放入与A对应的不同b和分数
2、循环遍历字典，将key加上前缀CB，value以从大到小的分数进行排序，并且相同的item以——分割，item和score间用：分割

#coding=utf-8
import sysinfile = '../data/cb.result'
outfile = '../data/cb_reclist.redis'ofile = open(outfile, 'w')MAX_RECLIST_SIZE = 100
PREFIX = 'CB_'rec_dict = {}
with open(infile, 'r') as fd:for line in fd:itemid_A, itemid_B, sim_score = line.strip().split('\t')#判断itemA在不在该字典里面，若不在，创建一个key为itemA的列表，把与itemA相关联的itemB和score添加进去if itemid_A not in rec_dict:rec_dict[itemid_A] = []rec_dict[itemid_A].append((itemid_B, sim_score))#循环遍历字典，格式化数据，把itemB和score中间以：分割，不同的itemB以_分割
for k, v in rec_dict.items():key_item = PREFIX + k#接下来格式化数据，将数据以从大到小排列后再格式化#排序,由于数据量大，我们只取100个#排好序后，我们来格式化数据reclist_result = '_'.join([':'.join([tu[0], str(round(float(tu[1]), 6))]) \for tu in sorted(v, key=lambda x: x[1], reverse=True)[:MAX_RECLIST_SIZE]])ofile.write(' '.join(['SET', key_item, reclist_result]))ofile.write("\n")ofile.close()

得到类似如下数据：根据itemid_a,返回相似对应itemid_b的score

SET CB_53051091
76 726100303:0.393048_953500302:0.393048_6193109237:0.348855

3.【召回】CF算法：

下文是利用Mapreduce实现CF算法，如果想利用Spark实现CF算法请参考的这篇文章https://blog.csdn.net/qq_36816848/article/details/113184759

（1）以userid itemid score形式整理训练数据

总体思路：首先和cb一样，对处理完的用户元数据，物品元数据，行为数据进行cf数据准备工作，我们的目的事输出： user，item
score，其中主要是的到用户对item的score，其中用户收听的音乐的时常和总的时长相除得到score

python gen_cf_train.py

#coding=utf-8
import sysinput_file = "../data/merge_base.data"# 输出cf训练数据
output_file = '../data/cf_train.data'
ofile = open(output_file, 'w')key_dict = {}
with open(input_file, 'r') as fd:for line in fd:ss = line.strip().split('\001')# 用户行为userid = ss[0].strip()itemid = ss[1].strip()watch_len = ss[2].strip()hour = ss[3].strip()# 用户画像gender = ss[4].strip()age = ss[5].strip()salary = ss[6].strip()user_location = ss[7].strip()# 物品元数据name = ss[8].strip()desc = ss[9].strip()total_timelen = ss[10].strip()item_location = ss[11].strip()tags = ss[12].strip()#拼接key，为了将同一个用户对相同物品的时长全部得到，需要做个聚合key = '_'.join([userid, itemid])if key not in key_dict:key_dict[key] = []key_dict[key].append((int(watch_len), int(total_timelen)))#循环处理相同用户对相同item的分数
for k, v in key_dict.items():t_finished = 0t_all = 0# 对<userid, itemid>为key进行分数聚合for vv in v:t_finished += vv[0]t_all += vv[1]# 得到userid对item的最终分数score = float(t_finished) / float(t_all)userid, itemid = k.strip().split('_')ofile.write(','.join([userid, itemid, str(score)]))ofile.write("\n")ofile.close()

得到如下数据：cf_train.data

userid, itemid, score 0189c9fecdd47bb64720c23a960272d3,935400252,1.3
014e7a8f4544bcd156365d3f348399c2,068800255,1.21889952153
00af96daaf12d1afa11d102f9f98fc3b,405100213,0.0581395348837
00383d3536ce00ad469cb1c57946686a,732009535,1.5703125
00fa3f43730a4374a43e1edfab614bb4,720400256,1.41509433962
0027834e40d613c175f715052aa341af,411500272,1.30735930736
00e5dd1b98a94f1976e49ffedc830e84,177200319,0.764705882353

（2）用ALS协同过滤算法跑出item-item数据 （套用协同过滤）
　　II矩阵数据准备，redis数据分为这么几个部分，这部分的数据需要利用到MapReduce框架，进行map和reduce排序。

一、MR实现CF算法

1.归一化， 归一化阶段我们主要是将相同的item进行单位模计算，把数据映射到0～1范围之内处理，更加便捷快速。因为我们要用到cos相似度计算公式，将相同的item的分数进行平方和再开根号，最后进行单位化。

相似度的计算公式:

2.取pair对 相同用户两两取pair，输出两次，形成II矩阵
3.计算总和 将相同pair的分数相加

（1）归一化：
　　1.map阶段，只要将转数据换成item，user，score ，因为我们要在reduce阶段进行相同item单位化，要充分用到shuffle阶段的排序。

#!usr/bin/python
# -*- coding: UTF-8 -*-
'''思路：转换成i，u，s的矩阵
'''
import  sysfor line in sys.stdin:ss = line.strip().split(',')if len(ss) != 3:continueu , i , s = ssprint '\t'.join([i,u,s])

2.reduce阶段，我们需要将相同item平方和相加开根号，然后再单位化计算，最后输出。

#!usr/bin/python
# -*- coding: UTF-8 -*-
'''在map的基础上将每个item进行归一化，map已经将相同的item排好序，根据map的结果进行给先平方再开根号：思路 ：1、截取字符串，取出item，user，socre2、在for循环中进行判断，当前的item和下一个是否相同，要是相同，将相同的放到列表（user，score）列表里面，否则往下执行3、若不相同，循环user和score列表，计算模计算，然后再次循环，进行单位化计算
'''import sys
import mathcur_item = None
user_score_list = []
for line in sys.stdin:ss = line.strip().split('\t')if len(ss) != 3:continueitem = ss[0]userid = ss[1]score = ss[2]#wordcount判断，当前和下一个是否相同，相同添加到列表，不相同进行归一化计算if cur_item == None:cur_item = itemif cur_item != item:#定义sumsum = 0.0#循环列表进行模向量计算for ss in user_score_list:user,s = sssum += pow(s,2)sum = math.sqrt(sum)#单位化计算for touple in user_score_list:u,s = touple# 进行单位化完成后，我们输出重置成原来的user-item-score输出print "%s\t%s\t%s" % (u, cur_item, float(s / sum))#初始化这两个变量cur_item = itemuser_score_list = []user_score_list.append((userid,float(score)))#定义sum
sum = 0.0
#循环列表进行模向量计算
for ss in user_score_list:user,s = sssum += pow(s,2)
sum = math.sqrt(sum)
#单位化计算
for touple in user_score_list:u,s = touple# 进行单位化完成后，我们输出重置成原来的user-item-score输出print "%s\t%s\t%s" % (u, cur_item, float(s / sum))

（2）两两取pair对

思路：两两取pair对，我们在map阶段，其实什么都不用做，保证输出user，itemid，score即可。**
map阶段：

#!usr/bin/python
# -*- coding: UTF-8 -*-#在进行pair取对之前，什么都不需要做，输出就行import  sysfor line in sys.stdin:u, i, s = line.strip().split('\t')print "%s\t%s\t%s" % (u, i, s)

reduce阶段：
　　将同一个用户下的item进行两两取对，因为要形成II矩阵，就必须以user为参考单位，相反形成uu矩阵，就必须以item参考，所以将同一个用户下的item进行两两取对，并将分数相乘，就得到临时这个相似度，因为还没有对相同pair对的分数相加，这个是最后一步要做的。

#!usr/bin/python
# -*- coding: UTF-8 -*-'''思路：进行map排好序之后，我们的会得到相同user对应的不同item和score，这里我们主要的思路是进行相同用户两两取pair1、进行判断，当前用户和下一个用户是不是一样，若是不一样，我们进行两两取对，形成ii矩阵2、若是相同，我们将不同的item和score放入list里面
'''import  syscur_user = None
item_score_list = []
for line in sys.stdin:user,item,score = line.strip().split('\t')if cur_user == None:cur_user= userif cur_user != user:#进行两两pair，利用range函数for i in range(0,len(item_score_list)-1):for j in range(i+1,len(item_score_list)):item_a, score_a = item_score_list[i]item_b, score_b = item_score_list[j]# 输出两遍的目的是为了形成II矩阵的对称print "%s\t%s\t%s" % (item_a, item_b, score_a * score_b)print "%s\t%s\t%s" % (item_b, item_a, score_a * score_b)cur_user = useritem_score_list = []item_score_list.append((item,float(score)))#进行两两pair，利用range函数
for i in range(0,len(item_score_list)-1):for j in range(i+1,len(item_score_list)):item_a, score_a = item_score_list[i]item_b, score_b = item_score_list[j]# 输出两遍的目的是为了形成II矩阵的对称print "%s\t%s\t%s" % (item_a, item_b, score_a * score_b)print "%s\t%s\t%s" % (item_b, item_a, score_a * score_b)

（3）进行最终分数求和，我们最后的阶段是要将相同pair的分数相加才能得到两个item的相似度。
map阶段，将相同item对排序到一起，就要将pair组成一个key进行排序，将同一个partition后数据放倒一个reduce桶中，MapReduce框架shuffle阶段，key只是做排序，partition只是做分区，不要搞混了。

#!usr/bin/python
# -*- coding: UTF-8 -*-'''sum的map中，我们需要把相同的itemA，itemB组成key，为了使相同的key能够在shuffle阶段分配到同一个reduce中，因为是计算item的相似度，要把相同的相加
'''import  sysfor line in sys.stdin:item_a,item_b,score = line.strip().split('\t')key = '#'.join([item_a,item_b])print '%s\t%s' %(key,score)
reduce阶段主要任务就是将相同的item的pair对相加.'''思路：将相同的item的分数进行相加，得到最后的相似度
'''import  syscur_item = None
score = 0.0
for line in sys.stdin:item, s = line.strip().split('\t')if not cur_item:cur_item = itemif cur_item != item:ss = item.split("#")if len(ss) != 2:continueitem_a, item_b = ssprint "%s\t%s\t%s" % (item_a, item_b, score)cur_item = itemscore = 0.0score += float(s)ss = item.split("#")
if len(ss) != 2:sys.exit()
item_a, item_b = ss
print "%s\t%s\t%s" % (item_a, item_b, score)

执行上述程序运行脚本run.sh

HADOOP_CMD="/usr/local/src/hadoop-2.6.5/bin/hadoop"
STREAM_JAR_PATH="/usr/local/src/hadoop-2.6.5/share/hadoop/tools/lib/hadoop-streaming-2.6.5.jar"#要想cf代码直接改成cf_train.data
INPUT_FILE_PATH_1="/cf_train.data"
OUTPUT_PATH_1="/output1"
OUTPUT_PATH_2="/output2"
OUTPUT_PATH_3="/output3"$HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH_1
$HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH_2
$HADOOP_CMD fs -rmr -skipTrash $OUTPUT_PATH_3Step 1.$HADOOP_CMD jar $STREAM_JAR_PATH \-input $INPUT_FILE_PATH_1 \-output $OUTPUT_PATH_1 \-mapper "python 1_gen_ui_map.py" \-reducer "python 1_gen_ui_reduce.py" \-jobconf "mapreduce.map.memory.mb=4096" \-file ./1_gen_ui_map.py \-file ./1_gen_ui_reduce.pyStep 2.$HADOOP_CMD jar $STREAM_JAR_PATH \-input $OUTPUT_PATH_1 \-output $OUTPUT_PATH_2 \-mapper "python 2_gen_ii_pair_map.py" \-reducer "python 2_gen_ii_pair_reduce.py" \-jobconf "mapreduce.map.memory.mb=4096" \-file ./2_gen_ii_pair_map.py \-file ./2_gen_ii_pair_reduce.pyStep 3.$HADOOP_CMD jar $STREAM_JAR_PATH \-input $OUTPUT_PATH_2 \-output $OUTPUT_PATH_3 \-mapper "python 3_sum_map.py" \-reducer "python 3_sum_reduce.py" \-jobconf "mapreduce.map.memory.mb=8000" \-file ./3_sum_map.py \-file ./3_sum_reduce.py
最后得到基于cf的ii矩阵
cf_train.data ，执行得到cf.result （最后一列没有超过1的）
000000228 006900337 0.495383099617
000000228 237400301 0.4655287556
000000228 489600256 0.327370227556
000000228 880800319 0.6522021568
000000228 895300223 0.0654423912424

（3）对数据格式化，item-> item list形式，整理出KV形式
代码:python gen_reclist.py
结果统一放入放入redis,读kv，以itemid_A为key,其余两列追加为value，放到rec_dict。
区分key,加前缀CB_ ，SET为redis命令，实现批量灌入

#coding=utf-8
'''思路：这个处理的逻辑和CB中完全一样，不一样的是redis的key是CF开头
'''import sysinfile = '../data/cf.result'
outfile = '../data/cf_reclist.redis'ofile = open(outfile, 'w')MAX_RECLIST_SIZE = 100
PREFIX = 'CF_'rec_dict = {}
with open(input_file,'r') as fd:for line in fd:itemid_A, itemid_B, score = line.strip().split('\t')#判断itemA在不在该字典里面，若不在，创建一个key为itemA的列表，把与itemA相关联的itemB和score添加进去if itemid_A not in rec_dict:rec_dict[itemid_A] = []rec_dict[itemid_A].append((itemid_B, score))#循环遍历字典，格式化数据，把itemB和score中间以：分割，不同的itemB以_分割
for k,v in rec_dict.items():key = PREFIX+k#接下来格式化数据，将数据以从大到小排列后再格式化#排序,由于数据量大，我们只取100个list = sorted(v,key=lambda x:x[1],reverse=True)[:MAX_RECLIST_SIZE]#拍好序后，我们来格式化数据result = '_'.join([':'.join([str(val[0]),str(round(float(val[1]),6))]) for val in list])ofile.write(' '.join(['SET',key,result]))ofile.write("\n")ofile.close()

四、灌库（redis）

1.Centos中安装redis，本文下载的是redis-2.8.3，下载对应安装包并进行源码编译(需要C编译yum install gcc-c++ )，先执行make，然后进入src目录中，得到bin文件（redis-server 服务器，redis-cli 客户端）
　　
2.启动redis server服务两种方法：
　　]# ./src/redis-server

3.导入数据

首先连接服务，换一个终端执行：]# ./src/redis-cli，连接redis。

**3.1 灌数据（批量灌）：**需要安装unix2dos进行格式转换（yum install unix2dos），安装完后执行unix2dos cb_reclist.reds命令。

再执行cat cb_reclist.redis | /usr/local/src/redis/redis-2.8.3/src/redis-cli --pipe
进入redis验证执行./src/redis-cli

获取数据并查看：127.0.0.1:6379> get CB_5305109176
726100303:0.393048_953500302:0.393048_6193109237:0.348855"

3.2 同上方法可以将cf灌库
　　unix2dos cf_reclist.redis
　　cat cf_reclist.redis | /usr/local/src/redis-2.8.3/src/redis-cli --pipe

4、LR训练模型的数据准备

准备我们自己的训练数据,其中标签label=watch_time / total_time
进入pre_data_for_rankmodel目录：python gen_samples.py，利用最开始的merge_base.data数据，最后输出samples.data。

思路：经过cb，cf算法，将数据已经放库，召回部分已经完成，接下来做排序模型，为逻辑回归准备样本数据
1、处理第一次将用户元数据，物品元数据，用户行为数据一起归并的数据，也就是merge_base.data,我们在这里需要得到用户画像数据，用户信息数据，标签数据
2、收取样本，标签，用户画像信息，物品信息
3、抽取用户画像信息，对性别和年龄生成样本数据
4、抽取item特征信息，分词获得token，score，做样本数据
5、拼接样本，生成最终的样本信息，作为模型进行训练 ‘’’

#coding=utf-8
import sys
sys.path.append('../')
reload(sys)
sys.setdefaultencoding('utf-8')import jieba
import jieba.analyse
import jieba.possegmerge_base_infile = '../data/merge_base.data'
output_file = '../data/samples.data'#我们这里需要再生成两个文件，一个是用户样本和item样本，因为要对实时推荐的化，必须使用这两个样本
output_user_feature_file = '../data/user_feature.data'
output_item_feature_file = '../data/item_feature.data'#这里生成个类似name和id对应的字典信息
output_itemid_to_name_file = '../data/name_id.dict'#定义函数，来获取各类数据
def get_base_samples(infile):#放待处理样本数据ret_samples_list = []#放user用户数据user_info_set = set()#放物品数据item_info_set = set()item_name2id = {}item_id2name = {}with open(infile, 'r') as fd:for line in fd:ss = line.strip().split('\001')if len(ss) != 13:continueuserid = ss[0].strip()itemid = ss[1].strip()#这两个时间为了计算label而使用watch_time = ss[2].strip()total_time = ss[10].strip()#用户数据gender = ss[4].strip()age = ss[5].strip()user_feature = '\001'.join([userid, gender, age])#物品数据name = ss[8].strip()item_feature = '\001'.join([itemid, name])#计算标签label = float(watch_time) / float(total_time)final_label = '0'if label >= 0.82:final_label = '1'elif label <= 0.3:final_label = '0'else:continue#接下来装在数据，并返回结果，首先我们装在itemid2name和itemname2iditem_name2id[name] = itemiditem_id2name[itemid] = name#装在待处理的标签数据ret_samples_list.append([final_label, user_feature, item_feature])user_info_set.add(user_feature)item_info_set.add(name)return ret_samples_list, user_info_set, item_info_set, item_name2id, item_id2name#step 1 程序的入口，开始调用函数，开始处理文件，得到相应的数据
base_sample_list, user_info_set, item_info_set, item_name2id, item_id2name = \get_base_samples(merge_base_infile)#step 2 抽取用户画像信息，用户标签转换，将年龄和age进行转换，用于样本使用
user_fea_dict = {}
for info in user_info_set:userid, gender, age = info.strip().split('\001')#设置标签idx，将男(1)和女(0)用数剧的形式表示，权重都设置为1idx = 0 # default 女if gender == '男':idx = 1#将标签和权重拼接起来gender_fea = ':'.join([str(idx), '1'])#性别设置完成，我们接下来设置年龄，将年龄进行划分，0-18，19-25，26-35，36-45idx = 0if age == '0-18':idx = 0elif age == '19-25':idx = 1elif age == '26-35':idx = 2elif age == '36-45':idx = 3else:idx = 4idx += 2age_fea = ':'.join([str(idx), '1'])user_fea_dict[userid] = ' '.join([gender_fea, age_fea])#step 3 抽取物品特征，这里我们要用到分词，将name进行分词，并且把分词后的token转换成id，这里就需要我们来做生成tokenid词表
token_set = set()
item_fs_dict = {}
for name in item_info_set:token_score_list = []for x,w in jieba.analyse.extract_tags(name,withWeight=True):token_score_list.append((x,w))token_set.add(x)item_fs_dict[name] = token_score_list#进行token2id的转换
token_id_dict = {}
#这里我们要用到刚刚利用set去重过的token列表，生成tokenid的字典表
for s in enumerate(list(token_set)):token_id_dict[s[1]] = s[0]#接下来，我们需要把第三步生成的item_fs_dict中name对应的token全部替换成id，然后当作字典，为下面的全量替换做准备
item_fea_dict = {}
user_feature_offset = 10
for name ,fea in item_fs_dict.items():token_score_list = []for (token,score) in fea:if token not in token_id_dict:continuetoken_id = token_id_dict[token] + user_feature_offsettoken_score_list.append(':'.join([str(token_id),str(score)]))#接下来输出到字典中item_fea_dict[name] = ' '.join(token_score_list)#step 4 将第一步输出的样本数据整体替换并且替换user_feature和item_feature,并输出到文件中
ofile = open(output_file,'w')
for (label,userfea,itemfea) in base_sample_list:userid = userfea.strip().split('\001')[0]item_name = itemfea.strip().split('\001')[1]if userid not in user_fea_dict:continueif item_name not in item_fea_dict:continueofile.write(' '.join([label,user_fea_dict[userid],item_fea_dict[item_name]]))ofile.write('\n')ofile.close()#step 5 为了能够实时使用userfeatre，我们需要输出一下
out_put_file = open(output_user_feature_file,'w')
for userid,fea in user_fea_dict.items():out_put_file.write('\t'.join([userid,fea]))out_put_file.write('\n')
out_put_file.close()#step 6 输出item_feature
out_file = open(output_item_feature_file,'w')
for name,fea in item_fea_dict.items():if name not in item_name2id:continueitemid = item_name2id[name]out_file.write('\t'.join([itemid,fea]))out_file.write('\n')#step 7 输出id2name的对应的字典
o_file = open(output_itemid_to_name_file,'w')
for id,name in item_id2name.items():o_file.write('\t'.join([id,name]))o_file.write('\n')
o_file.close()

得到如下数据：

1.item_feature.data

2.user_feature.data

3.samples.data（将1、2拼接）
由userid、item_name拼接

五、模型准备

lr.py

思路：用到数据，需要写load_data的部分，
1.定义main，方法入口，然后进行load_data的编写
2.调用该方法的到x训练x测试，y训练，y测试，使用L1正则化或是L2正则化使得到结果更加可靠
3. 输出wegiht，和b偏置

# -*- coding: UTF-8 -*-import sys
import numpy as np
from scipy.sparse import csr_matrixfrom sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegressioninput_file = sys.argv[1]def load_data():#由于在计算过程用到矩阵计算，这里我们需要根据我们的数据设置行，列，和训练的数据准备#标签列表target_list = []#行数列表fea_row_list = []#特征列表fea_col_list = []#分数列表data_list = []#设置行号计数器row_idx = 0max_col = 0with open(input_file,'r') as fd:for line in fd:ss = line.strip().split(' ')#标签label = ss[0]#特征fea = ss[1:]#将标签放入标签列表中target_list.append(int(label))#开始循环处理特征：for fea_score in fea:sss = fea_score.strip().split(':')if len(sss) != 2:continuefeature, score = sss#增加行fea_row_list.append(row_idx)#增加列fea_col_list.append(int(feature))#填充分数data_list.append(float(score))if int(feature) > max_col:max_col = int(feature)row_idx += 1row = np.array(fea_row_list)col = np.array(fea_col_list)data = np.array(data_list)fea_datasets = csr_matrix((data, (row, col)), shape=(row_idx, max_col + 1))x_train, x_test, y_train, y_test = train_test_split(fea_datasets, s, test_size=0.2, random_state=0)return x_train, x_test, y_train, y_testdef main():x_train,x_test,y_train,y_test = load_data()#用L2正则话防止过拟合model = LogisticRegression(penalty='l2')#模型训练model.fit(x_train,y_train)ff_w = open('model.w', 'w')ff_b = open('model.b', 'w')#写入训练出来的Wfor w_list in model.coef_:for w in w_list:print >> ff_w, "w: ", w# 写入训练出来的Bfor b in model.intercept_:print >> ff_b, "b: ", bprint "precision: ", model.score(x_test, y_test)print "MSE: ", np.mean((model.predict(x_test) - y_test) ** 2)if __name__ == '__main__':main()

六、推荐系统实现

推荐系统demo流程：
初始化工作：加载data，获取user_feature.data，item_feature.data特征 ，用字典维护。
（1）解析请求：userid，itemid
（2）加载模型：加载排序模型（model.w，model.b）
（3）检索候选集合：利用cb，cf去redis里面检索数据库，得到候选集合
（4）获取用户特征user_feature.data ：userid
（5）获取物品特征item_feature.data ：itemid
（6）打分(逻辑回归函数sigmoid: 1 / (1 + exp(-wx)))，排序
（7）top-n过滤（精排）
（8）数据包装（itemid->name），返回

执行主函数 main.py

#coding=utf-8
import web
import sys
import redis
import json
import mathurls = ('/', 'index','/test', 'test',
)app = web.application(urls, globals())# 加载user特征
user_fea_dict = {}
with open('../data/user_feature.data') as fd:for line in fd:userid, fea_list_str = line.strip().split('\t')user_fea_dict[userid] = fea_list_str# 加载item特征
item_fea_dict = {}
with open('../data/item_feature.data') as fd:for line in fd:ss = line.strip().split('\t')if len(ss) != 2:continueitemid, fea_list_str = ssitem_fea_dict[itemid] = fea_list_strclass index:def GET(self):r = redis.Redis(host='master', port=6379,db=0)# step 1 : 解析请求，上面我们已经得到userid，itemidparams = web.input()userid = params.get('userid', '')req_itemid = params.get('itemid', '')# step 2 : 加载模型model_w_file_path = '../rankmodel/model.w'model_b_file_path = '../rankmodel/model.b'model_w_list = []model_b = 0.with open (model_w_file_path, 'r') as fd:for line in fd:ss = line.strip().split(' ')if len(ss) != 3:continuemodel_w_list.append(float(ss[2].strip()))with open (model_b_file_path, 'r') as fd:for line in fd:ss = line.strip().split(' ')model_b = float(ss[2].strip())# step 3 : 检索候选(match)，这里我们分两次，cb，cf#将检索回来的item全部放到recallitem列表里面rec_item_mergeall = []# 3.1 cfcf_recinfo = 'null'key = '_'.join(['CF', req_itemid])if r.exists(key):cf_recinfo = r.get(key)if len(cf_recinfo) > 6:for cf_iteminfo in cf_recinfo.strip().split('_'):item, score = cf_iteminfo.strip().split(':')rec_item_mergeall.append(item)# 3.2 cbcb_recinfo = 'null'key = '_'.join(['CB', req_itemid])if r.exists(key):cb_recinfo = r.get(key)if len(cb_recinfo) > 6:for cb_iteminfo in cb_recinfo.strip().split('_'):item, score = cb_iteminfo.strip().split(':')rec_item_mergeall.append(item)# step 4: 获取用户特征,将获取的用户特征处理后放到字典里面，方便后续计算内积user_fea = ''if userid in user_fea_dict:user_fea = user_fea_dict[userid]u_fea_dict = {}for fea_idx in user_fea.strip().split(' '):ss = fea_idx.strip().split(':')if len(ss) != 2:continueidx = int(ss[0].strip())score = float(ss[1].strip())u_fea_dict[idx] = score# step 5: 获取物品的特征 ,循环遍历刚刚得到itemid，判断item是否在item特征中，若在开始进行处理rec_list = []for itemid in rec_item_mergeall:if itemid in item_fea_dict:item_fea = item_fea_dict[itemid]i_fea_dict = dict()for fea_idx in item_fea.strip().split(' '):ss = fea_idx.strip().split(':')if len(ss) != 2:continueidx = int(ss[0].strip())score = float(ss[1].strip())i_fea_dict[idx] = score#得到召回item对应的特征和用户的特征，之后根据模型求出来的w，b，进行打分wx_score = 0.#这里我们求个内积，wx，然后做sigmoid，先将两个字典拼接起来，然后计算分数for fea, score in dict(u_fea_dict.items() + i_fea_dict.items()).items():wx_score += (score * model_w_list[fea])#**计算sigmoid: 1 / (1 + exp(-wx))**final_rec_score = 1 / (1 + math.exp(-(wx_score + model_b)))#将itemid和分数放入列表中，方便后续排序rec_list.append((itemid, final_rec_score))# step 6 : 精排序(rank)rec_sort_list = sorted(rec_list, key=lambda x:x[1], reverse=True)# step 7 : 过滤(filter)取top10rec_fitler_list = rec_sort_list[:10]# step 8 : 返回+包装(return)，进行将itemid转换成nameitem_dict = {}with open('../data/name_id.dict', 'r') as fd:for line in fd:raw_itemid, name = line.strip().split('\t')item_dict[raw_itemid] = nameret_list = []for tup in rec_fitler_list:req_item_name = item_dict[req_itemid]item_name = item_dict[tup[0]]item_rank_score = str(tup[1])ret_list.append('   ->   '.join([req_item_name, item_name, item_rank_score]))ret = '\n'.join(ret_list)return retclass test:def GET(self):print web.input()return '222'if __name__ == "__main__":app.run()

七.推荐系统测试：

保证redis启动，执行python main.py 9999
(py27) [root@master rec_server]# 192.168.179.10:9999/?userid=00370d83b51febe3e8ae395afa95c684&itemid=3880409156
浏览器输入上述返回推荐top10。

可视化页面查看：进入web目录执行 python page_web.py 9999
输入useid_itemid