假如有一些用戶(hù)對(duì)物品的評(píng)分?jǐn)?shù)據(jù)，怎么預(yù)測(cè)用戶(hù)的喜好并給他推送感興趣的物品呢？我們常常能想到的一種成熟方法就是協(xié)同過(guò)濾，這里只談?wù)劵谟脩?hù)相似度的協(xié)同過(guò)濾。原理很好理解，物以類(lèi)聚，你感興趣的物品很可能是和你相似的人喜歡的物品。

如下表是一個(gè)用戶(hù)-物品評(píng)分矩陣，大寫(xiě)的"A","B","C"表示用戶(hù)（user），小寫(xiě)的"a","b","c"表示物品(item)，數(shù)字表示對(duì)應(yīng)評(píng)分(rating)。缺的地方表示用戶(hù)還沒(méi)對(duì)物品進(jìn)行評(píng)分，我們需要估計(jì)用戶(hù)對(duì)這些沒(méi)評(píng)過(guò)分的物品的大致評(píng)分，將評(píng)分高的物品推薦給他。假設(shè)A,B,C互相相似，相似度為1，那么我們可以得=算出A對(duì)b預(yù)估評(píng)分=（2*1+3*1）/（1+1）=2.5分。B對(duì)c預(yù)估評(píng)分3.5，依此類(lèi)推。

真正的推薦系統(tǒng)里，數(shù)據(jù)量巨大，如新聞資訊類(lèi)推薦，用戶(hù)有1w個(gè)，新聞?dòng)?0w條，我們難以在內(nèi)存里計(jì)算這一過(guò)程，所以下面探究如何利用大數(shù)據(jù)工具（如spark）來(lái)實(shí)現(xiàn)推薦。

引入spark等相關(guān)包。

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

構(gòu)造我們的模擬數(shù)據(jù)ratings，轉(zhuǎn)成spark里的dataframe格式，然后轉(zhuǎn)成rdd并將數(shù)據(jù)按用戶(hù)進(jìn)行g(shù)roup。

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)

用戶(hù)*用戶(hù)作笛卡爾積，得到
[(('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)]))]
計(jì)算用戶(hù)兩兩之間相似度

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位小數(shù)
        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)])]
當(dāng)然這里用共同評(píng)分的項(xiàng)算cosine相似度，假設(shè)共同評(píng)分的項(xiàng)如果只有1個(gè)，那么相似度就一定為1。這是不合理的。所以具體算相似度可以設(shè)置min_common_item_bar為一個(gè)大一點(diǎn)的值，保證只有較多公共評(píng)分的項(xiàng)時(shí)才能算相似度，或者用jaccard等相似度計(jì)算方法。

#將相似用戶(hù)作為key
def f(x): return x
temp = user_sim_rdd.flatMapValues(f).map(lambda x:(x[1][0], (x[0], x[1][1])))

#join相似用戶(hù)和用戶(hù)對(duì)新聞評(píng)分
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)

這里進(jìn)行一些格式的變換（flatten等）將數(shù)據(jù)變成
[('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)]
這里的每一條數(shù)據(jù)（user, item, rating, similarity）是說(shuō)與user相似度為similarity的某個(gè)用戶(hù)對(duì)item的評(píng)分為rating。所以預(yù)測(cè)user對(duì)未知項(xiàng)item的評(píng)分，可以通過(guò)所有(user, item,,)的評(píng)分和相似度來(lái)做加權(quán)平均得到一個(gè)估計(jì)值，這里我沒(méi)找到更好方法直接對(duì)rdd操作，所以先轉(zhuǎn)為dataframe，用sql來(lái)計(jì)算。

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")

最終得到評(píng)分
+-------+-------+---------------+
|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|
+-------+-------+---------------+
當(dāng)然這里把所有已知評(píng)分也預(yù)測(cè)了評(píng)分，其實(shí)應(yīng)該把這部分剔除掉的。最后對(duì)user推薦的就是該user的weighted_rating較高的item了。還有就是做加權(quán)平均的時(shí)候可以取similarity最高的前k個(gè)，不用像這里一樣拿所有來(lái)算。或者先對(duì)用戶(hù)進(jìn)行聚類(lèi)，加權(quán)評(píng)分這個(gè)用戶(hù)所在類(lèi)的用戶(hù)評(píng)分。