提到ALS相信大家應該都不會覺得陌生（不陌生你點進來幹嘛[捂臉]），它是協同過濾的一種，并被內建到Spark的Mllib庫中。本文就ALS的基本原理進行講解，并手把手、肩并肩地帶您實作這一算法。
- 原理篇
我們用人話而不是大段的數學公式來講講ALS是怎麼一回事。
1.1 你聽說過推薦算法麼
假如我是豆瓣的CEO，很多豆瓣的使用者在豆瓣電影上都會對電影進行評分。那麼根據這個評分資料，我們有可能知道這些使用者除了自己評過分的電影之外還喜歡或讨厭哪些電影嗎？這就是一個典型的推薦問題，解決這一類問題的算法被稱為推薦算法。
1.2 什麼是協同過濾
協同過濾的英文全稱是Collaborative Filtering，簡稱CF。注意，這不是一款遊戲！從字面上分析，協同就是尋找共同點，過濾就是篩選出優質的内容。
1.3 協同過濾的分類
一般來說，協同過濾推薦分為三種類型：
- 基于使用者(user-based)的協同過濾，通過計算使用者和使用者的相似度找到跟使用者A相似的使用者B, C, D...再把這些使用者喜歡的内容推薦給A；
- 基于物品(item-based)的協同過濾，通過計算物品和物品的相似度找到跟物品1相似的物品2, 3, 4...再把這些物品推薦給看過物品1的使用者們；
- 基于模型(model based)的協同過濾。主流的方法可以分為：矩陣分解，關聯算法，聚類算法，分類算法，回歸算法，神經網絡。
1.4 矩陣分解
矩陣分解 (decomposition, factorization)是将矩陣拆解為數個矩陣的乘積。比如豆瓣電影有m個使用者，n個電影。那麼使用者對電影的評分可以形成一個m行n列的矩陣R，我們可以找到一個m行k列的矩陣U，和一個k行n列的矩陣I，通過U * I來得到矩陣R。
1.5 ALS
如果想通過矩陣分解的方法實作基于模型的協同過濾，ALS是一個不錯的選擇，其英文全稱是Alternating Least Square，翻譯過來是交替最小二乘法。假設使用者為a，物品為b，評分矩陣為R(m, n)，可分解為使用者矩陣U(k, m)和物品矩陣I(k, n)，其中m, n, k代表矩陣的次元。前方小段數學公式低能預警：
根據矩陣分解的定義，有
用MSE作為損失函數，為了友善化簡，加法符号左側的常數改為-1/2
- 對損失函數求U_a的一階偏導數，那
協同過濾(ALS)的原理及Python實作 - 令一階偏導數等于0
協同過濾(ALS)的原理及Python實作 - 同理，可證
協同過濾(ALS)的原理及Python實作
1.6 求解使用者矩陣U和物品矩陣I
矩陣R是已知的，我們随機生成使用者矩陣U，
- 利用1.5中的式5、R和U求出I
- 利用1.5中的式6、R和I求出U
如此交替地執行步驟1和步驟2，直到算法收斂或者疊代次數超過了最大限制，最終我們用RMSE來評價模型的好壞。
- 實作篇
本人用全宇宙最簡單的程式設計語言——Python實作了ALS算法，沒有依賴任何第三方庫，便于學習和使用。簡單說明一下實作過程，更詳細的注釋請參考本人github上的代碼。
注：代碼中用到的Matrix類是我寫的一個矩陣類，可以取出矩陣的行或列，計算矩陣的乘法、轉置和逆。
2.1 建立ALS類
初始化，存儲使用者ID、物品ID、使用者ID與使用者矩陣列号的對應關系、物品ID與物品矩陣列号的對應關系、使用者已經看過哪些物品、評分矩陣的Shape以及RMSE。
class ALS(object):
def __init__(self):
self.user_ids = None
self.item_ids = None
self.user_ids_dict = None
self.item_ids_dict = None
self.user_matrix = None
self.item_matrix = None
self.user_items = None
self.shape = None
self.rmse = None 2.2 資料預處理
對訓練資料進行處理，得到使用者ID、物品ID、使用者ID與使用者矩陣列号的對應關系、物品ID與物品矩陣列号的對應關系、評分矩陣的Shape、評分矩陣及評分矩陣的轉置。
def _process_data(self, X):
self.user_ids = tuple((set(map(lambda x: x[0], X))))
self.user_ids_dict = dict(map(lambda x: x[::-1],
enumerate(self.user_ids)))
self.item_ids = tuple((set(map(lambda x: x[1], X))))
self.item_ids_dict = dict(map(lambda x: x[::-1],
enumerate(self.item_ids)))
self.shape = (len(self.user_ids), len(self.item_ids))
ratings = defaultdict(lambda: defaultdict(int))
ratings_T = defaultdict(lambda: defaultdict(int))
for row in X:
user_id, item_id, rating = row
ratings[user_id][item_id] = rating
ratings_T[item_id][user_id] = rating
err_msg = "Length of user_ids %d and ratings %d not match!" % (
len(self.user_ids), len(ratings))
assert len(self.user_ids) == len(ratings), err_msg
err_msg = "Length of item_ids %d and ratings_T %d not match!" % (
len(self.item_ids), len(ratings_T))
assert len(self.item_ids) == len(ratings_T), err_msg
return ratings, ratings_T 2.3 使用者矩陣乘以評分矩陣
實作稠密矩陣與稀疏矩陣的矩陣乘法，得到使用者矩陣與評分矩陣的乘積。
def _users_mul_ratings(self, users, ratings_T):
def f(users_row, item_id):
user_ids = iter(ratings_T[item_id].keys())
scores = iter(ratings_T[item_id].values())
col_nos = map(lambda x: self.user_ids_dict[x], user_ids)
_users_row = map(lambda x: users_row[x], col_nos)
return sum(a * b for a, b in zip(_users_row, scores))
ret = [[f(users_row, item_id) for item_id in self.item_ids]
for users_row in users.data]
return Matrix(ret) 2.4 物品矩陣乘以評分矩陣
實作稠密矩陣與稀疏矩陣的矩陣乘法，得到物品矩陣與評分矩陣的乘積。
def _items_mul_ratings(self, items, ratings):
def f(items_row, user_id):
item_ids = iter(ratings[user_id].keys())
scores = iter(ratings[user_id].values())
col_nos = map(lambda x: self.item_ids_dict[x], item_ids)
_items_row = map(lambda x: items_row[x], col_nos)
return sum(a * b for a, b in zip(_items_row, scores))
ret = [[f(items_row, user_id) for user_id in self.user_ids]
for items_row in items.data]
return Matrix(ret) 2.5 生成随機矩陣
def _gen_random_matrix(self, n_rows, n_colums):
data = [[random() for _ in range(n_colums)] for _ in range(n_rows)]
return Matrix(data) 2.6 計算RMSE
def _get_rmse(self, ratings):
m, n = self.shape
mse = 0.0
n_elements = sum(map(len, ratings.values()))
for i in range(m):
for j in range(n):
user_id = self.user_ids[i]
item_id = self.item_ids[j]
rating = ratings[user_id][item_id]
if rating > 0:
user_row = self.user_matrix.col(i).transpose
item_col = self.item_matrix.col(j)
rating_hat = user_row.mat_mul(item_col).data[0][0]
square_error = (rating - rating_hat) ** 2
mse += square_error / n_elements
return mse ** 0.5 2.7 訓練模型
資料預處理
變量k合法性檢查
生成随機矩陣U
交替計算矩陣U和矩陣I，并列印RMSE資訊，直到疊代次數達到max_iter
儲存最終的RMSE
def fit(self, X, k, max_iter=10):
ratings, ratings_T = self._process_data(X)
self.user_items = {k: set(v.keys()) for k, v in ratings.items()}
m, n = self.shape
error_msg = "Parameter k must be less than the rank of original matrix"
assert k < min(m, n), error_msg
self.user_matrix = self._gen_random_matrix(k, m)
for i in range(max_iter):
if i % 2:
items = self.item_matrix
self.user_matrix = self._items_mul_ratings(
items.mat_mul(items.transpose).inverse.mat_mul(items),
ratings
)
else:
users = self.user_matrix
self.item_matrix = self._users_mul_ratings(
users.mat_mul(users.transpose).inverse.mat_mul(users),
ratings_T
)
rmse = self._get_rmse(ratings)
print("Iterations: %d, RMSE: %.6f" % (i + 1, rmse))
self.rmse = rmse 2.8 預測一個使用者
預測一個使用者感興趣的内容，剔除使用者已看過的内容。然後按感興趣分值排序，取出前n_items個内容。
def _predict(self, user_id, n_items):
users_col = self.user_matrix.col(self.user_ids_dict[user_id])
users_col = users_col.transpose
items_col = enumerate(users_col.mat_mul(self.item_matrix).data[0])
items_scores = map(lambda x: (self.item_ids[x[0]], x[1]), items_col)
viewed_items = self.user_items[user_id]
items_scores = filter(lambda x: x[0] not in viewed_items, items_scores)
return sorted(items_scores, key=lambda x: x[1], reverse=True)[:n_items] 2.9 預測多個使用者
循環調用2.8，預測多個使用者感興趣的内容。
def predict(self, user_ids, n_items=10):
return [self._predict(user_id, n_items) for user_id in user_ids] 3 效果評估
3.1 main函數
使用電影評分資料集，訓練模型并統計RMSE。
@run_time
def main():
print("Tesing the accuracy of ALS...")
X = load_movie_ratings()
model = ALS()
model.fit(X, k=3, max_iter=5)
print()
print("Showing the predictions of users...")
user_ids = range(1, 5)
predictions = model.predict(user_ids, n_items=2)
for user_id, prediction in zip(user_ids, predictions):
_prediction = [format_prediction(item_id, score)
for item_id, score in prediction]
print("User id:%d recommedation: %s" % (user_id, _prediction)) 3.2 效果展示
設定k=3，疊代5次，并展示了前4個使用者的推薦内容，最終RMSE為0.370，運作時間46.5秒，效果還算不錯~
3.3 工具函數
本人自定義了一些工具函數，可以在github上檢視：
https://github.com/tushushu/imylu/tree/master/imylu/utils- run_time - 測試函數運作時間
- load_movie_ratings - 加載電影評分資料
總結
ALS的原理：雞生蛋、蛋生雞
ALS的實作：基本上就是矩陣乘法
原文釋出時間為：2019-1-2
本文作者：李小文
本文來自雲栖社群合作夥伴“
Python愛好者社群”，了解相關資訊可以關注“python_shequ”微信公衆号