假如有一些用戶對物品的評分數(shù)據(jù)贮竟，怎么預(yù)測用戶的喜好并給他推送感興趣的物品呢台丛？我們常常能想到的一種成熟方法就是協(xié)同過濾箍邮，這里只談?wù)劵谟脩粝嗨贫鹊膮f(xié)同過濾。原理很好理解糙申，物以類聚宾添，你感興趣的物品很可能是和你相似的人喜歡的物品船惨。
如下表是一個用戶-物品評分矩陣柜裸，大寫的"A","B","C"表示用戶（user），小寫的"a","b","c"表示物品(item)粱锐，數(shù)字表示對應(yīng)評分(rating)疙挺。缺的地方表示用戶還沒對物品進行評分，我們需要估計用戶對這些沒評過分的物品的大致評分怜浅，將評分高的物品推薦給他铐然。假設(shè)A,B,C互相相似，相似度為1恶座，那么我們可以得=算出A對b預(yù)估評分=（2*1+3*1）/（1+1）=2.5分搀暑。B對c預(yù)估評分3.5，依此類推跨琳。
| 用戶/物品 | a | b | c |
|---|---|---|---|
| A | 2.0 | 2.0 | |
| B | 1.0 | 2.0 | |
| C | 3.0 | 5.0 |
真正的推薦系統(tǒng)里自点，數(shù)據(jù)量巨大，如新聞資訊類推薦脉让，用戶有1w個桂敛，新聞有20w條，我們難以在內(nèi)存里計算這一過程溅潜，所以下面探究如何利用大數(shù)據(jù)工具（如spark）來實現(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ù)按用戶進行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)
用戶*用戶作笛卡爾積设捐，得到
[(('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位小數(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)])]
當然這里用共同評分的項算cosine相似度潦牛，假設(shè)共同評分的項如果只有1個，那么相似度就一定為1挡育。這是不合理的巴碗。所以具體算相似度可以設(shè)置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等）將數(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）是說與user相似度為similarity的某個用戶對item的評分為rating召噩。所以預(yù)測user對未知項item的評分，可以通過所有(user, item,,)的評分和相似度來做加權(quán)平均得到一個估計值逸爵，這里我沒找到更好方法直接對rdd操作具滴，所以先轉(zhuǎn)為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|
+-------+-------+---------------+
當然這里把所有已知評分也預(yù)測了評分构韵，其實應(yīng)該把這部分剔除掉的。最后對user推薦的就是該user的weighted_rating較高的item了趋艘。還有就是做加權(quán)平均的時候可以取similarity最高的前k個疲恢，不用像這里一樣拿所有來算〈呻剩或者先對用戶進行聚類显拳，加權(quán)評分這個用戶所在類的用戶評分。
