美文网首页python机器学习爬虫
使用Spark运行协同过滤推荐算法

使用Spark运行协同过滤推荐算法

作者: JianlingZhou | 来源:发表于2018-07-15 00:28 被阅读81次

接近一个月没写博客,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万条记录上跑完协同过滤算法,这个过程只能说又锻炼了一遍我配置环境的能力,看着这么大的数据在集群里这么短时间跑完还是很开心的。

结果

相关文章

网友评论

    本文标题:使用Spark运行协同过滤推荐算法

    本文链接:https://www.haomeiwen.com/subject/iywspftx.html