一、项目需求

1、背景需求

工业方案的推荐，由用户上传的文章，类似百度文库，存在一定的免费与付费文章。

2、数据描述

产品已经运行了一年多，现在需读其进行完善，加入推荐系统，

业务的同事直接将mysql业务库中，文章表、评价表、文章标签维度表，导出到本地，形成 articles.csv，ratings.csv，tags.csv文件，

我们将读取数据并加载到 mongodb 和 elasticsearch 中。

• 1）方案文章数据集：有 10 个字段，每个字段之间通过“^”符号进行分割。如:方案ID、名称、页数、描述、语言、类型、原作者、上传者、上传时间

• 2）用户评分数据集：用户UID， 方案mid、分数score、时间戳timestamp

• 3）方案标签数据集：用户UID， 方案mid、标签score、时间戳timestamp

3、项目框架

mongodb-linux-x86_64-rhel62-3.4.3 
redis-4.0.2
 
<spark.version>2.1.1</spark.version>
<kafka.version>0.10.2.1</kafka.version>
<redis.version>2.9.0</redis.version>
<mongodb-spark.version>2.0.0</mongodb-spark.version>

<elasticsearch-spark.version>6.4.3</elasticsearch-spark.version>
<elasticsearch.version>6.4.3</elasticsearch.version>
1
2
3
4
5
6
7
8
9
10

二、相关代码

1、本地文件写入Mongo与ES

• 加载对应路径下的数据 得到RDD
• 将 RDD --> DataFrame
• 将DataFrame 保存到Mongo
• 将DataFrame保存到ES

1） 加载对应路径下的数据 得到RDD

2）将 RDD --> DataFrame

反射机制，定义一个样例类，后期直接映射成DataFrame的schema信息；

3）将DataFrame 保存到Mongo

4）将DataFrame保存到ES

方案数据集，保存到ES，便于搜索

2、基于统计推荐

0）数据源使用评分数据

1）历史热门统计

读取评分数据集，统计所有评分中评分数最多的文章，从大到小排序，结果写入 MongoDB 的相应数据集中。

2）最近热门文章：

读取评分数据集，通过 UDF 函数将评分的数据时间修改为月， 然后统计每月文章的评分数。结果写入 MongoDB 的相应数据集中。

3）平均得分统计：

读取评分数据集，通过执行 SQL 语 句实现对于文章的平均分统计

4）每个类别优质文章统计：

在刚刚计算完整个文章的平均得分之后，将文章集合与文章类型做笛卡尔积，然后 过滤掉文章类型不符合的条目，

join，给文章数据集添加一列《平均值。

fillter：判断，当前文章是否包含该类型，如果不包含，就过滤。
第一个map：将数据转换为，key为文章类型，value为当前文章类型下的文章列表（文章ID，文章平均分）
第二个map：groupbykey后，将数据转换为存储的格式，key为文章类型，value为当前文章类型下的文章列表（文章ID，文章平均分），根据list的特性，进行倒序sort，取前10个。

3、基于内容推荐

内容推荐概述

内容推荐（Content-based Recommendations ,CB）: 根据物品的内在特征，计算物品间的相似度，推荐相似的物品。

物品内在特征提取方法；

• 1.打标签法：专家标签、用户自定义标签、降维分析数据，提取隐语义标签(LFM)
• 2.关键词法（文本信息特征提取）：语义处理和情感分析(NLP)、潜在语义分析(LSA)

相似度计算方法： 欧式距离 ，余弦相似度

本项目采用：

文本特征提取主要： count ，tfidf(词频，一个词在一个文本出现很多，在其他文本出现很少，说明它具有很好的分类能力， 如每篇文章都会出现“工业”这个标签，这个标签特征的分类能力就不强，但是一个标签为“继电器”，这个标签在其他文章出现的频率很低，就具有很好的标签。

tfidf，越普遍，值越小

实现流程：

TF-IDF：提取文章的特征
余弦相似度，计算文章的相似度，

• 1）加载数据：加载文章数据, 数据预处理, 文章类别genres列，竖线替换为空格

• 2）TF-IDF特征值计算

  • (1）tokennizer分词，利用spark自带的分词器 tokennizer，将genres列分成单个词形成array，添加列words
  • (2）词频TF计算，利用spark自带的词频计算对象hashingTF，计算words列每个词的 TF词频，添加列rawFeatures；
  • (3）IDF逆文档计算，利用spark自带的词频计算模型idfModel，计算rawFeatures中 TF词频的IDF逆文档，添加列features；
  • TF-IDF后，便有了特征

• 3）余弦相似度计算，得到推荐列表

  • (1）文章数据中，两两文章做笛卡尔积，特征值相似度计算
  • (2）每篇文章，过滤掉自己和相似度小于0.65的，倒序排序，取前十个作为该文章的推荐文章列表

0）数据源为文章数据集

1）TF-IDF特征计算

内容特征：文章类型，一篇文章genres列包含多个类型，用空格分开，

需要对文章的不同类别，做特征提取、转为特征向量，再计算两两特征向量相似值（余弦夹角），降序排列，选取前10个作为备选列表。

（1）分词：

（2）得到TF词频(Term Frequency)

（50,[11,13,19,43,49],[1.0,1.0,1.0,1.0,1.0]）
50：设置的特征数
11：adventure标签的索引编号，
1：代表该在对应的索引位置，的词频为1
是一个稀疏矩阵

（3）得到IDF是逆文本频率指数(Inverse Document Frequency)。

2）余弦相似度计算

（1） features 特征值稀疏矩阵， 转换为一个array，Double类型

（2）两两文章特征值相似度计算，笛卡尔积，

4、基于协同过滤推荐

目前推荐系统主要分为三类：
(1)内容推荐CB，主要根据 物品本身的内容特征，发现物品或者内容的相关性，通过物品的相似性推荐。喜欢动漫，就推荐动漫相关的书。

(2) 协同过滤推荐CF， 不仅考虑物品的相似性，还考虑用户的相似性，计算用户的相似度；根据用户对物品或者信息的偏好，发现物品或者内容本身的相关性，或者是发现用户的相关性；喜欢动漫的人也喜欢探险类的书，就可以给另一个喜欢动漫的人也推荐探险类的书。

(3) 混合推荐：该技术采用一些策略（如加权、切换、特征放大等整合多种推荐技术，比如基于内容的推荐系统，基于人口统计学的推荐系统，基于效用的推荐系统和基于基础知识的推荐系统。

协同推荐概述：

1）基于用户的协同过滤：根据所有用户对物品或者信息的偏好，发现与当前用户口味和偏好相似的“邻居”用户群，在一般的应用中是采用计算“K-邻居”的算法。然后，基于这K个邻居的历史偏好信息，为当前用户进行推荐。

假设用户1喜欢物品A、C，用户2喜欢物品B，用户3喜欢物品 A 、C和物品D；
可以发现，用户1和用户3的口味和偏好是比较类似的，
同时用户3还喜欢物品D，那么我们可以推断用户1可能也喜欢物品D，
因此可以将物品D推荐给用户1。

2）基于项目的协同过滤：它使用所有用户对物品或者信息的偏好，发现物品和物品之间的相似度，然后根据用户的历史偏好信息，将类似的物品推荐给用户。

假设用户1喜欢物品A、C，用户2喜欢物品A、B、C，用户 3喜欢物品A，
可以分析出，物品A和物品C时比较类似的，喜欢物品A的人都喜欢物品C，
基于这个数据可以推断，用户3很有可能也喜欢物品 C，
所以系统会将物品C推荐给用户3。

物品数据基本稳定 -> item-CF -> 电商
物品数据不稳定，用户特性不怎么变 -> user-CF -> 新闻推荐，用户画像

而基于模型的方法，是要使用这些偏好数据来训练模型，找到内在规律，再用模型来做预测（类
似回归)

3）基于模型的协同过滤，主要是用机器学习的思想来建模解决， 基于样本的用户偏好信息，训练一个推荐模型，找到内在规律，再用模型来做预测（类似回归)。 包括关联算法、神经网络、矩阵分解、聚类算法、分类算法、回归算法、图模型以及隐语义模型。

• 矩阵分解：目前用矩阵分解做协同过滤是目前使用也很广泛的一种方法。因为在协同过滤使用过程中经常会遇到数据稀疏性的问题，通过矩阵分解将高维矩阵分解为低维矩阵再进行计算。目前主流的矩阵分解推荐算法主要是SVD的一些变种，比如FunkSVD，BiasSVD和SVD++。

训练模型时，可以基于标签内容来提取物品特征，也可以让模型去发掘物品的潜在特征；这样的模型被称为隐语义模型(Latent Factor Model，LFM)

• 基于概率的隐语义分析（pLSA)
• 隐式迪利克雷分布模型( LDA)
• 矩阵因子分解模型(基于奇异值分解的模型,SVD)

通过矩阵分解，进行降维分析

• 协同过滤算法非常依赖历史数据，而一般的推荐系统中，偏好数据又往往是稀疏的:这就需要对原始数据做降维处理。
• 分解之后的矩阵,就代表了用户和物品的隐藏特征

矩阵分解模型(奇异值分解,SVD)

评分矩阵为R，现在有m个用户,n个物品
想要发现k个隐类，就是找到两个矩阵P和Q，使这两个矩阵的乘积近似等于R，

即，将用户物品评分矩阵R，分解成为两个低维矩阵相乘:

0）实现流程

• 1）用户评分数据 ，转换为spark自带的Rating数据集(uid,mid,score);
• 2）根据Rating数据集， 使用spark自带的ALS交替最小二乘模型，训练隐语义模型;
• 3）将uid 数据和 mid数据 做笛卡尔积，产生(uid，mid) 空矩阵 ;
• 4）使用训练后的模型，预测(uid，mid) 空矩阵，得到预测评分矩阵;
• 5）将预测结果通过预测分值进行排序。
• 6）返回分值最大的 K 个文章，作为当前用户的推荐。

通过用户评分稀疏矩阵，奇异值分解，矩阵的相乘，
训练预测后得到：
预测评分矩阵，包含该用户对未打分法物品的预测评分，实现推荐。
同时得到：
用户隐藏特征矩阵
物品隐藏特征矩阵，用于实时推荐。

1）数据源，从评分数据集中提取出的

2） 转换为Rating评分矩阵， 通过ALS训练隐语义模型

3） 模型预测，获取评分最高的50个文章

4） 得到 含有隐特征的文章特征矩阵Q，为实时推荐做准备

5、实时推荐

实现流程

用户点评一个文章，就会更新一下对于他的推荐列表，

• 对当前文章，打高分>=3 ，会推荐该文章隐含特征高度相似的文章
• 对当前文章，打低分 ，会减少推荐该文章隐含特征高度相似的文章，推荐他历史评分高的相似文章。

数据采集

• 1）用户实时评分时，产生评分后台日志（与rating数据一样uid id score time）；
• 2）flume实时采集评分日志sink到kafka；

spark sreaming计算
加载mongodb中电影相似度矩阵数据，把它广播出去

• 1）spark 读取矩阵分解得到的文章隐特征矩阵，进行广播，相当于维度信息。
• 2）spark 读取kafka中用户评分流数据 ，.foreachRDD进行流计算，实时更新推荐列表；
  • 1，流处理时，先从redis里获取当前用户最近的20次评分，保存成Array[(mid, score)]
  • 2、从文章隐特征矩阵，得到当前电影最相似的20个文章，形成备选文章列表
  • 3、将备选文章列表，与用户最近评分过的文章的评分做加权, 对每个备选文章，计算推荐优先级，得到当前用户的实时推荐列表

由于评分数据为热点数据，为了保证效率，降低延迟，会将用户最近评分数据存储到redis；

MongoDB中，该评分文章的相似隐藏特征的相似文章，在隐语义推荐时，通过model.productFeature，提前计算好了， 降低延时。

默认端口号：redis 6379，mongodb 27017 ，mysql 3306，

实时推荐算法

当用户 u 对文章 p 进行了评分，将触发一次对 u 的推荐结果的更新。

由于用户u对文章 p 评分，对于用户 u来说，他与p最相似的文章们之间的推荐强度将发生变化，所以选取与文章 p 最相似的 K 个文章作为候选文章。

• 每个候选文章按照“推荐优先级”这一权重，作为衡量这个文章被推荐给用户 u 的优先级。
• 这些文章将根据用户 u 最近的若干评分，计算出各自对用户 u 的推荐优先级；
• 然后与上次对用户 u 的实时推荐结果，进行基于推荐优先级的合并、替换得到更新后 的推荐结果。
  
  （该用户最近评分的文章20个，与 ，当前评分高相似文章20个，的相似度和，再乘以该用户对当前文章的评分），除以20，

具体来说:

• 首先，获取用户 u 按时间顺序最近的 K 个评分，记为 RK；获取文章 p 的最相似的 K 个文章集合，记为 S;
• 然后，对于每个文章 q，计算其推荐优先级 ，计算公式如下:

• 把相似度和评分结合起来做一个加权操作先拿到最近的K个评分, 另外拿到和当前评分的电影相似的电影作为备选,
• 有一个备选电影列表接下来考察每一个备选电影, 按照模型给一个推荐评分,就是把备选电影和最近评分过的电影挨个做比对,
• 他俩是不是可以从相似度矩阵中取出他们的相似度, 他俩的相似度再乘以最近评分过的电影的评分是不是就相当于做了加权,
• 最后求和之后再除以评分过的个数, 是不是做了一个加权的平均数啊, 这就是我们最后推荐分数的基准项, 最后还有两个偏移项, 分别是奖励和惩罚

1、数据源：电影相似度矩阵数据、kafka评分流

2、流处理，算法实现

1). 从redis里获取当前用户最近的K次评分，保存成Array[(mid, score)]

2). 从相似度矩阵中取出当前电影最相似的N个电影，作为备选列表，Array[mid]

3). 将备选电影列表和最近评分过的电影的评分做加权, 对每个备选电影，计算推荐优先级，得到当前用户的实时推荐列表，Array[(mid, score)]

4). 把推荐数据保存到mongodb