假如有一些用户对物品的评分数据，怎么预测用户的喜好并给他推送感兴趣的物品呢？我们常常能想到的一种成熟方法就是协同过滤，这里只谈谈基于用户相似度的协同过滤。原理很好理解，物以类聚，你感兴趣的物品很可能是和你相似的人喜欢的物品。

如下表是一个用户-物品评分矩阵，大写的"A","B","C"表示用户（user），小写的"a","b","c"表示物品(item)，数字表示对应评分(rating)。缺的地方表示用户还没对物品进行评分，我们需要估计用户对这些没评过分的物品的大致评分，将评分高的物品推荐给他。假设A,B,C互相相似，相似度为1，那么我们可以得=算出A对b预估评分=（2*1+3*1）/（1+1）=2.5分。B对c预估评分3.5，依此类推。

真正的推荐系统里，数据量巨大，如新闻资讯类推荐，用户有1w个，新闻有20w条，我们难以在内存里计算这一过程，所以下面探究如何利用大数据工具（如spark）来实现推荐。

引入spark等相关包。

from pyspark import SparkConf, SparkContext
from pyspark.sql import HiveContext

#python packages
import numpy as np
import time

import pandas as pd
from pyspark.sql.types import *
from pyspark.ml.recommendation import ALS
from pyspark.sql.functions import col
import numpy as np
import json
from operator import itemgetter

构造我们的模拟数据ratings，转成spark里的dataframe格式，然后转成rdd并将数据按用户进行group。

ratings = [('A', 'a', 2.0), ('A', 'c', 2.0), ('B', 'a', 1.0), ('B', 'b', 2.0), ('C', 'b', 3.0), ('C', 'c', 5.0)]
mySchema = StructType([ StructField("user_id", StringType(), True) ,StructField("news_id", StringType(), True), StructField("rating", DoubleType(), True)])
ratings = spark.createDataFrame(ratings, schema=mySchema)

rating_by_user_rdd = ratings.rdd.map(lambda x: (x.user_id, (x.news_id, x.rating))).groupByKey().mapValues(list)

得到格式
[('A', [('a', 2.0), ('c', 2.0)]),
('B', [('a', 1.0), ('b', 2.0)]),
('C', [('b', 3.0), ('c', 5.0)])]

rating_cross = rating_by_user_rdd.cartesian(rating_by_user_rdd)

用户*用户作笛卡尔积，得到
[(('A', [(1, 2.0), (3, 2.0)]), ('A', [(1, 2.0), (3, 2.0)])),
(('A', [(1, 2.0), (3, 2.0)]), ('B', [(1, 1.0), (2, 2.0)])),
(('B', [(1, 1.0), (2, 2.0)]), ('A', [(1, 2.0), (3, 2.0)])),
(('B', [(1, 1.0), (2, 2.0)]), ('B', [(1, 1.0), (2, 2.0)])),
(('A', [(1, 2.0), (3, 2.0)]), ('C', [(2, 3.0), (3, 5.0)])),
(('B', [(1, 1.0), (2, 2.0)]), ('C', [(2, 3.0), (3, 5.0)])),
(('C', [(2, 3.0), (3, 5.0)]), ('A', [(1, 2.0), (3, 2.0)])),
(('C', [(2, 3.0), (3, 5.0)]), ('B', [(1, 1.0), (2, 2.0)])),
(('C', [(2, 3.0), (3, 5.0)]), ('C', [(2, 3.0), (3, 5.0)]))]
计算用户两两之间相似度

min_common_item_bar = 1

# 1.user_based CF. user_similarity
#common_item < min_common_item_bar则return None
def user_sim_func(xi):
    l1 = xi[0][1]
    l2 = xi[1][1]
    common_item = set([kv[0] for kv in l1]).intersection(set([kv[0] for kv in l2]))
    if len(common_item) >= min_common_item_bar:
        vector_1 = [kv[1] for kv in l1 if kv[0] in common_item]
        vector_2 = [kv[1] for kv in l2 if kv[0] in common_item]
        #保留3位小数
        cos = np.around(cos_sim(vector_1, vector_2),5)
        return (xi[0][0],  (xi[1][0], cos))


def cos_sim(vector_a, vector_b):
    vector_a = np.mat(vector_a)
    vector_b = np.mat(vector_b)
    num = float(vector_a * vector_b.T)
    denom = np.linalg.norm(vector_a) * np.linalg.norm(vector_b)
    cos = num / denom
    return cos


user_sim_rdd = rating_cross.map(user_sim_func).filter(lambda x: x is not None).groupByKey().mapValues(list)

得到
[('B', [('A', 1.0), ('B', 1.0), ('C', 1.0)]),
('C', [('A', 1.0), ('B', 1.0), ('C', 1.0)]),
('A', [('A', 1.0), ('B', 1.0), ('C', 1.0)])]
当然这里用共同评分的项算cosine相似度，假设共同评分的项如果只有1个，那么相似度就一定为1。这是不合理的。所以具体算相似度可以设置min_common_item_bar为一个大一点的值，保证只有较多公共评分的项时才能算相似度，或者用jaccard等相似度计算方法。

#将相似用户作为key
def f(x): return x
temp = user_sim_rdd.flatMapValues(f).map(lambda x:(x[1][0], (x[0], x[1][1])))

#join相似用户和用户对新闻评分
temp2 = temp.join(rating_by_user_rdd)

def prepare_rating(x):
    similar_user = x[0]
    target_user = x[1][0][0]
    similarity = float(x[1][0][1])
    articles = x[1][1] 
    
    res = []
    for article in articles:
        res.append((target_user, article[0], article[1], similarity))
        
    return res
    
temp3 = temp2.map(prepare_rating).flatMap(f)

这里进行一些格式的变换（flatten等）将数据变成
[('B', 'a', 2.0, 1.0),
('B', 'c', 2.0, 1.0),
('C', 'a', 2.0, 1.0),
('C', 'c', 2.0, 1.0),
('A', 'a', 2.0, 1.0),
('A', 'c', 2.0, 1.0),
('B', 'a', 1.0, 1.0),
('B', 'b', 2.0, 1.0),
('C', 'a', 1.0, 1.0),
('C', 'b', 2.0, 1.0)]
这里的每一条数据（user, item, rating, similarity）是说与user相似度为similarity的某个用户对item的评分为rating。所以预测user对未知项item的评分，可以通过所有(user, item,,)的评分和相似度来做加权平均得到一个估计值，这里我没找到更好方法直接对rdd操作，所以先转为dataframe，用sql来计算。

mySchema = StructType([ StructField("user_id", StringType(), True) ,StructField("news_id", IntegerType(), True), StructField("rating", DoubleType(), True), StructField("similarity", DoubleType(), True)])
for_rating1 = temp3.toDF(schema=mySchema)

for_rating1.createOrReplaceTempView("for_rating1")
predict_rating = spark.sql("SELECT user_id, news_id, SUM(rating*similarity)/SUM(similarity) AS weighted_rating FROM for_rating1 GROUP BY user_id, news_id")

最终得到评分
+-------+-------+---------------+
|user_id|news_id|weighted_rating|
+-------+-------+---------------+
| C| 3| 3.5|
| A| 3| 3.5|
| A| 1| 1.5|
| A| 2| 2.5|
| C| 1| 1.5|
| B| 1| 1.5|
| B| 3| 3.5|
| C| 2| 2.5|
| B| 2| 2.5|
+-------+-------+---------------+
当然这里把所有已知评分也预测了评分，其实应该把这部分剔除掉的。最后对user推荐的就是该user的weighted_rating较高的item了。还有就是做加权平均的时候可以取similarity最高的前k个，不用像这里一样拿所有来算。或者先对用户进行聚类，加权评分这个用户所在类的用户评分。