项目实战系列|个性化广告推荐系统实战系列（六）（实时推荐产生结果）实时推荐|pyspark|redis|个性化广

1. 写在前面这几天打算整理一个模拟真实情景进行广告推荐的一个小Demon，这个项目使用的阿里巴巴提供的一个淘宝广告点击率预估的数据集，采用lambda架构，实现一个离线和在线相结合的实时推荐系统，对非搜索类型的广告进行点击率预测和推荐(没有搜索词，没有广告的内容特征信息)。这个感觉挺接近于工业上的那种推荐系统了，通过这个推荐系统，希望能从工程的角度了解推荐系统的流程，也顺便学习一下大数据的相关技术，这次会涉及到大数据平台上的数据处理，离线处理业务和在线处理业务，涉及到的技术包括大数据的各种技术，包括Hadoop，Spark(Spark SQL, Spqrk ML, Spark-Streaming)， Redis，Hive，HBase，Kafka和Flume等，机器学习的相关技术（数据预处理，模型的离线训练和在线更新等。所以这几天的时间借机会走一遍这个流程，这里也详细记录一下，方便以后回看和回练，这次的课程是跟着B站上的一个课程走的，讲的挺详细的，就是没有课件和资料，得需要自己搞，并且在实战这次的推荐系统之前，最好是有一整套的大数据环境(我已经搭建好了），然后就可以来玩这个系统了哈哈，现在开始
上一篇文章又走了一波召回，把每个用户感兴趣的500篇广告和用户，广告自身的特征已经缓存到了redis中去，这里我们来体验一波实时推荐的流程，也就是把逻辑回归模型加载到内存，通过服务接口的形式暴露出去。这样，来个用户id和资源位id，我就可以先拿着用户id到数据库里面找到他感兴趣的500个广告(上面召回到redis里面了)，有了广告，再从缓存里面拿到广告的相关特征和该用户的特征，这样就构成了逻辑回归模型原始特征，我们再按照之前的处理方式转成one-hot，拼接成特征向量，然后就可以跑逻辑回归模型了预测点击率，按照这个套路，把500个广告的点击率排个序，然后就可以把点击率高的推荐给用户了。最后，把这这个代码封装起来，就产生了第一篇文章里面的那个效果了。
逻辑比较清晰，开启三台虚拟机，Hadoop集群，spark集群和redis服务，jupyter notebook，现在开始。
2. 实时推荐产生结果 【项目实战系列|个性化广告推荐系统实战系列（六）（实时推荐产生结果）】这里面的核心操作就是实时数据预处理成逻辑回归可用数据的过程。首先这里面我们需要先导入几个映射关系字典，就是各种离散特征one-hot时候所提到的原始特征与one-hot里面1的位置对应关系字典。因为我们这里数据预处理的主要操作就是one-hot的部分，我们拿到用户的特征之后，相应的类别数值要转成one-hot的形式，必须依赖我们当时训练模型时候的one-hot对应表才行，具体可以看第三篇文章。形式类似于这样的：

文章图片

这里就直接拿来对应了，代码如下：

# pvalue_level pvalue_level_rela = { -1: 0, 3:3, 1:2, 2:1}# new_user_class_level new_user_class_level_rela = { -1:0, 3:2, 1:4, 4:3, 2:1}# cms_group_id cms_group_id_rela = {7: 9, 11: 6, 3: 0, 8: 8, 0: 12, 5: 3, 6: 10, 9: 5, 1: 7, 10: 4, 4: 1, 12: 11, 2: 2 }# final_gender_code final_gender_code_rela = { 1:1, 2:0}# age_level age_level_rela = { 3:0, 0:6, 5:2, 6:5, 1:4, 4:1, 2:3}# shopping_level shopping_level_rela = { 3:0, 1:2, 2:1}occupation_rela = { 0:0, 1:1}pid_rela = {"430548_1007": 0, "430549_1007": 1 }

下面就是最关键的核心操作，逻辑是这样，首先函数里面的接收值是用户id和广告位id，拿到用户id之后，我先从redis的10号数据库把用户的特征拿出来，这里用的json导入hget()函数，这是个字典的形式。然后再基于这个用户id，从9号数据库拿到该用户的广告召回列表，然后遍历列表里面的每个广告，从10号数据库里面拿到广告的特征。然后选择出之前逻辑回归用到的那几个特征，进行one-hot编码，这里会用到上面字典里面的对应关系，根据拿到特征的数值映射出1的位置了，然后补上1得到。最终用DenseVector把所有特征进行合并，存入最终的列表中去， 500个都遍历完毕之后，返回列表。具体代码如下，注意这个代码里面埋了个雷，编程的粗心导致，最后报错的时候在指出来。

host = "192.168.56.101" port = 6379def create_datasets(userId, pid):client_of_recall = redis.StrictRedis(host=host, port=port, db=9) client_of_features = redis.StrictRedis(host=host, port=port, db=10)# 获取用户特征 user_feature = json.loads(client_of_features.hget("user_features1", userId))# 获取用户召回集 recall_sets = client_of_recall.smembers(userId)result = [] # 遍历召回集 for adgroupId in recall_sets: adgroupId = int(adgroupId) # 获取该广告的特征值 ad_feature = json.loads(client_of_features.hget("ad_features", adgroupId))features = { } features.update(user_feature) features.update(ad_feature)for k, v in features.items(): if v is None:# 如果特征为空，用-1补上 features[k] = -1features_col = [ # 特征值 "price", "cms_group_id", "final_gender_code", "age_level", "shopping_level", "occupation", "pid", "pvalue_level", "new_user_class_level" ] ''' "cms_group_id", 类别型特征，约13个分类 ==> 13维 "final_gender_code", 类别型特征，2个分类 ==> 2维 "age_level", 类别型特征，7个分类 ==>7维 "shopping_level", 类别型特征，3个分类 ==> 3维 "occupation", 类别型特征，2个分类 ==> 2维 '''# 下面把类别特征转成one-hot price = float(features["price"])pid_value = https://www.it610.com/article/[0 for i in range(2)]# [0, 0] cms_group_id_value = [0 for i in range(13)] final_gender_code_value = [0 for i in range(2)] age_level_value = [0 for i in range(7)] shopping_level_value = [0 for i in range(3)] occupation_value = [0for i in range(2)] pvalue_level_value = [0 for i in range(4)] new_user_class_level_value = [0 for i in range(5)]# 首先是生成所有的0，然后根据对应字典把1的位置补上 pid_value[pid_rela[pid]] = 1 cms_group_id_value[cms_group_id_rela[int(features["cms_group_id"])]] = 1 final_gender_code_value[final_gender_code_rela[int(features["final_gender_code"])]] = 1 age_level_value[age_level_rela[int(features["age_level"])]] = 1 shopping_level_value[shopping_level_rela[int(features["shopping_level"])]] = 1 occupation_value[occupation_rela[int(features["occupation"])]] = 1 pvalue_level_value[pvalue_level_rela[int(features["pvalue_level"])]] = 1 new_user_class_level_value[new_user_class_level_rela[int(features["new_user_class_level"])]] = 1vector = DenseVector([price] + pid_value + cms_group_id_value + final_gender_code_value + age_level_value \ + age_level_value + shopping_level_value + occupation_value + pvalue_level_value + new_user_class_level_value)result.append((userId, adgroupId, vector))return result

这段代码看起来很长，但是逻辑不是太难，就是上面的那段话。下面我们基于一个用户，指定广告位来看看结果：

文章图片

下面两行代码把上面这个准备成逻辑回归数据集的形式，即转成spark的DataFrame。

pdf = pd.DataFrame(create_datasets(100036, "430548_1007"), columns=["userId", "adgroupId", "features"]) datasets = spark.createDataFrame(pdf) datasets.show(5)

看眼结果：

文章图片

这样我们的实时数据准备完毕，下面就是导入之前训练好的CTRModel_AllOneHot模型来进行预测即可。

from pyspark.ml.classification import LogisticRegressionModel CTR_model = LogisticRegressionModel.load("hdfs://master:9000/user/icss/RecommendSystem/model/CTRModel_AllOneHot.obj")prediction = CTR_model.transform(datasets).sort("probability")

这里会迎来第一个报错，就是features_1报错找不到，这是因为之前训练的逻辑回归模型，用的列名叫做features_1，这个在第四篇里面。

文章图片

所以这里会报错找不到这个特征，于是我们这里必须把datasets里面的feature列改名成feature_1。

datasets = datasets.withColumnRenamed("features",'features1')

这样再执行预测的代码，报错

Caused by: java.lang.IllegalArgumentException: requirement failed: BLAS.dot(x: Vector, y:Vector) was given Vectors with non-matching sizes: x.size = 46, y.size = 39

文章图片

这个错看原因是size匹配不上，这个原因是model训练的时候用的维度是39维的特征，到了这里传入的是个46维的特征，所以大小匹配不上。这个就是说的上面的那个雷，在这里：

文章图片

这个特征正好是一个7维的one-hot向量。所以这里删掉一个，然后重新运行上面的代码就OK了，得到的预测结果如下：

文章图片

下面我们推荐前20个广告：

文章图片

到这里就基本上介绍完了，后面最终的那个效果就不实现了，那个就是写一个.py文件，把上面这些都封装到里面去，然后输入用户id和广告位id，然后最终的这个推荐结果即可。
3. 总结这里需要总结一下子了，关于这篇文章就不总结了，比较简单，下面总结一下这个项目。
花了大约一周的时间走完了这个小demo项目，这个项目可谓是麻雀虽小，但是涉及到技术和知识还是挺多的，并且把推荐系统里面最重要的两大模块召回和排序都给涉及到了。这个是基于天池的一个真实广告数据集模拟了一个简单的真实情景下的推荐系统，既涉及到了数据分析和处理，算法，又涉及到了大数据和分布式，走完之后感觉学习到的知识还是挺多的，至少对分布式下的推荐流程和数据的处理方式有了一个了解。当然，如果让我闭着眼重新写遍这个代码，我写不出来，毕竟只是通一遍根本没法学会pyspark的相关数据处理方式啊，函数模型啥的，只能说明白个大概，知道大致上有这么个东西，后期如果真用到了，不会陌生且能够快速上手，感觉这样这几天就是有意义的。另外，通过第三篇里面遇到的各种坑，感觉心理承受能力确实增强，现在对于报错反而有种小激动了哈哈。
至于这6篇文章，也是环环紧扣，从第一篇一步一步走的，完全是跟着视频里面的流程，亲自实践，然后用类似日记的形式整理下来，就是为了方便将来回看回练或者查起来方便，也希望能通过这几篇文章帮助更多像我一样推荐系统的小白学习者。具体内容的话就不串了，重点就是召回，数据预处理(one-hot，缺失值填充那一套)，然后是排序模型的一套。关于召回，还有很多种方式，像协同过滤，基于内容等，这个在之前新闻推荐的比赛里面涉及的很全，可以到那里看。具体到时候用哪种召回方式，根据自己的业务场景来。而排序这块，涉及到的模型更是好多，逻辑回归只是最简单的一种方式，还有很多深度学习模型。只不过这个在分布式的环境中怎么使用还是个迷(我目前不知道咋用)，所以这个也是我下一步要攻克的目标，这个小demon还有没涉及到的日志搜集数据和消息队列等使用，也是我下一步要攻克的目标，对的，下一波会玩一个黑马头条的推荐大项目，这个初步听了一下非常有意思，涉及的东西更加偏工程性，会从真实情景中收集数据，然后处理，离线和在线推荐，且会用到TensorFlow搭建深度学习模型，依然分布式环境，依然是大数据推荐，这个下一步会走一遍，感兴趣的伙伴可以一起，到时候遇到问题还能互相讨论。
当然这个我目前先不打算整理成博客，毕竟这东西要大很多，和这个小demon不是一个量级，整理起来超级费时间，并且基于搭建的大数据环境和这几篇的试错，大数据环境这边试错也差不多了感觉，最重要的是这段时间面临找工作，推荐这边除了工程，算法更加难啃，感觉后期会非常忙。等啥时候找着工作了，啥时候再梳理分享哈哈。
最后整理一个知识点作为结束： SparkML和SparkMLlib的区别：

spark mllib 基于RDD
- 数据准备需要创建一个基于LabeledPoint的RDD
- LabeledPoint（目标，[特征]）
- 已经停止更新了处于维护状态
spark ML 基于dataframe
- 数据准备需要把所有的特征放到一列中 dataframe还需要有一列是目标值
- model = lr.setLabelCol(‘affairs’).setFeaturesCol(‘feautures’).fit(trainDF)
- spark ML 与 sklearn更类似
- 最新的API放到 spark ML中的