接近一个月没写博客,6月的课程很繁忙,各种Presentation,各种考试,各种课程报告,期末考完了,写篇博客记录下之前的一个作业。
研究生期间有门课,大数据分析与决策。这门课有个作业是做个Presentation,实现基于矩阵分解的推荐算法,要求在包含超过1000万条数据的数据集上运行。
数据集
数据不难搞到,老师直接指定MovieLens数据集。MovieLens是一个含有1700万条电影评分数据的数据集,评分数据文件rating.csv里面的数据格式很简单:一个二维表,表头 userId,movieId,rating。
然后,1700多万行评分记录,整个csv文件都有几百兆,我用 vim 打开都卡了一下,用 VS Code打开更是卡的不行。
算法
算法,矩阵分解。就是把评分矩阵R(n个用户m个电影)分解成两个矩阵乘积。U矩阵是用户的特征矩阵,V矩阵是物品的特征矩阵。
如果R矩阵里的每个评分都是已知的,那么这个矩阵分解起来就比较容易,但是现在问题是R矩阵里面大多数评分是空缺的,UV就不好分解了。这个时候可以定义一个误差函数,通过一个最优化过程来实现分解。至于预测结果,就通过UV乘积的某行某列去查就可以了。
因为之前完全没有接触过推荐算法,也没有接触过Spark,这个题目乍一看比较吓人,其实也没有多吓人。重大研究生的水平老师们都清楚,一门课而已,并非真的叫你把算法重新实现一遍,直接调库就好了。其实实现一个单机版本也没有多难,把误差函数定义好,梯度下降方法也不难写,但是考虑到:
- 评分矩阵很大很稀疏,我粗略的算了一下,直接把评分矩阵(13Wx3W)在内存里展开,需要 30GB,算上分解出来的矩阵,可能需要64G内存的电脑,远远超出32bit程序的内存限制
- 最优化的参数太多,想要收敛到最小不知道会耗费多少精力
所以直接调Spark的mllib的ALS算法,因为Spark的API不是很熟,代码也是抄的网上的:
# coding: utf-8
import sys
from os.path import join
from pyspark.sql import SparkSession
from pyspark.sql import Row
from pyspark.mllib.recommendation import ALS
reload(sys)
sys.setdefaultencoding("utf-8")
# 训练模型的时候只使用了 ratings.csv 文件
def train_model(training, num_iterations=10, rank=1, lambda_=0.01):
return ALS.train(training.rdd, iterations=num_iterations, rank=rank, lambda_=lambda_, seed=0)
def recommend_for_all(model, movies, result_path, file_out_flag=False):
def parse_recommendations(line):
res = []
for item in line[1]:
movie_name = b_movies.value[item[1]]
res.append((movie_name, float(item[2])))
return Row(user=line[0], recommendations=res)
b_movies = spark.sparkContext.broadcast(dict((int(l[0]), l[1]) for l in movies.collect()))
products_for_all_users = model.recommendProductsForUsers(10).map(parse_recommendations).toDF()
recommendation_result = products_for_all_users.repartition(1).orderBy(products_for_all_users.user).rdd \
.map(lambda l: Row(str(l.user) + "," + str(list((r[0], r[1]) for r in l.recommendations))))\
.toDF().repartition(1)
if file_out_flag: # 输出到文件
recommendation_result.write.text(result_path)
def recommend_for_users(model, movies, user_ids):
def recommend_for_one_user(model, user_id):
b_movies = spark.sparkContext.broadcast(dict((int(l[0]), l[1]) for l in movies.collect()))
recommendations = model.recommendProducts(user_id, 10)
result = []
for item in recommendations:
result.append((b_movies.value[item.product], item.rating))
print "Recommendations for user ", user_id, ": ", result
if isinstance(user_ids, list):
for user in user_ids:
recommend_for_one_user(model, user)
elif isinstance(user_ids, int):
recommend_for_one_user(model, user_ids)
def main(spark, data_path, result_path):
movies = spark.read.csv(join(data_path, "movies.csv")).rdd.map(
lambda l: Row(int(l[0]), l[1], l[2])).toDF(["movieId", "title", "genres"])
ratings = spark.read.csv(join(data_path, "ratings.csv")).rdd.map(
lambda l: Row(int(l[0]), int(l[1]), float(l[2]))).toDF(["userId", "movieId", "rating"])
all_users = ratings.select("userId").distinct()
all_movies = ratings.select("movieId").distinct()
print "Got %d ratings from %d users on %d movies." \
% (ratings.count(), all_users.count(), all_movies.count())
# 划分训练集和测试集
training, test = ratings.randomSplit([0.8, 0.2], 1)
model = train_model(training)
# 为以下所有用户作出推荐并输出到终端
#user_ids = list(u.userId for u in all_users.take(3))
#print "userIds: ", user_ids
#recommend_for_users(model, movies, user_ids)
# 为所有的用户作出推荐并输出到文件
file_out_flag = True
recommend_for_all(model, movies, result_path, file_out_flag)
import os
if __name__ == "__main__":
if len(sys.argv) == 3:
data_path = sys.argv[1] # 数据目录
result_path = sys.argv[2] # 保存结果目录
else:
print "Usage: $SPARK_HOME/bin/spark-submit --master spark://master:7077 " \
"movieLens.py /data_path/ /result/path"
sys.exit()
spark = SparkSession \
.builder \
.appName("MovieLens Recommendation") \
.getOrCreate()
real_data_path = os.environ.get('SPARK_HOME') + '/' + data_path
main(spark, real_data_path, result_path)
spark.stop()
以上程序要求机器有Python2.7,装了numpy:
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple numpy
部署
部署稍微讲究一下,因为没跑过千万级的数据,我一开始就搭建了一个Spark集群。这个Spark集群没有运行在 hdfs 上,所以我需要把数据集文件拷到每台机器上,而且文件的位置必须一模一样,不然Python脚本读文件会出错,这里我把数据集放到了Spark目录下面(通过环境变量拿路径)。
关于Spark环境的搭建,因为我没有上Hadoop,使用的也是默认的集群管理器,所以环境搭建很简单。先确保Java环境搭好,然后把Spark下载下来解压到/usr/share/spark,之后设置环境变量SPARK_HOME,把$SPARK_HOME$/bin,$SPARK_HOME$/sbin添加到PATH里就好了。
然后启动Master结点:
start-master.sh -h 0.0.0.0 -p 7077
再启动Slave结点:
start-slave.sh spark://masterIP:7077
这个时候访问 http://masterIP:8080 可以查看集群状态:
我在这里用了4台Ubuntu系统的电脑,搭了一个有23GB内存,13个核的集群。然后我在任何一台有Python脚本的电脑上用以下命令提交任务到 Master 上:
spark-submit --master spark://masterIP:7077 movieLens.py ml-20m/ result/
# movieLens.py 是代码,后面两个参数是Python的__main__()的参数。
这个配置的集群基本7分钟左右就可以在1700万条记录上跑完协同过滤算法,这个过程只能说又锻炼了一遍我配置环境的能力,看着这么大的数据在集群里这么短时间跑完还是很开心的。
网友评论