基于 Spark RDD 的电影点评数据分析

学习 Spark 的基本用法，利用 Spark 的 map、sort、join、reduce 等功能，对电影点评数据进行简单的分析。

初始化 Spark Context

Python

import findspark

# 初始化 spark 环境，系统全局环境变量已经设置 SPARK_HOME, PYSPARK_PYTHON 则不需要给此函数传入环境变量。
findspark.init()

from pyspark.conf import SparkConf
from pyspark.sql import SparkSession

appname = "rdd movies - 1. top 10 - user_47"  # 任务名称（将 user_xx 替换成你的号码）
master = "yarn"  # spark 集群地址

# spark 资源配置
conf = SparkConf().setAppName(appname).setMaster(master)
spark = SparkSession.builder.config(conf=conf).getOrCreate()

# 获取作业上下文
sc = spark.sparkContext

准备数据

Python

# 读取 HDFS 文件
movies = sc.textFile("hdfs://node1:8020/text/ml_1m/movies.dat")
ratings = sc.textFile("hdfs://node1:8020/text/ml_1m/ratings.dat")
users = sc.textFile("hdfs://node1:8020/text/ml_1m/users.dat")

Python

# 过滤异常数据
movies = movies.filter(
    lambda movie: (len(movie.strip()) > 0) and (len(movie.split("::")) == 3)
)
movies.cache()
ratings = ratings.filter(
    lambda rating: (len(rating.strip()) > 0) and (len(rating.split("::")) == 4)
)
ratings.cache()
users = users.filter(
    lambda user: (len(user.strip()) > 0) and (len(user.split("::")) == 5)
)
users.cache()

输出所有电影中评分最高的前 10 个电影名称和平均评分

Python

# 定义函数，输入【电影 ID 和评分】、【电影 ID 和电影名称】和【评分前 n 位】，输出【评分最高的电影名称和平均评分】
def top(ratings_data, movies_data, n):
    # 从 ratings_data 中提取出【电影 ID】和【评分】，并在 value 中补充“1”作为【评分数量】
    movieID_rating = ratings_data.map(
        lambda x: (int(x.split("::")[1]), (int(x.split("::")[2]), 1))
    )
    # 对同一部电影，加总【评分】和【评分数量】，得到【评分和】和【评分数量和】
    movieID_rating_reduced = movieID_rating.reduceByKey(
        lambda x, y: (x[0] + y[0], x[1] + y[1])
    )
    # 对同一部电影，用【评分和】除以【评分数量和】，得到【平均评分】
    movieID_rating_average = movieID_rating_reduced.mapValues(lambda x: x[0] / x[1])
    # 根据每一部电影的【平均评分】从大到小排序
    movieID_rating_descending = movieID_rating_average.sortBy(lambda x: x[1], False)
    # 提取【平均评分】最高的 n 部电影的【电影 ID】和【平均评分】，并存入一个 RDD
    movieID_rating_top_n = sc.parallelize(movieID_rating_descending.take(n))
    # 从 movies_data 中提取出【电影 ID】和【电影名称】
    movieID_movieName = movies_data.map(
        lambda x: (int(x.split("::")[0]), x.split("::")[1])
    )
    # 将【电影名称】合并到【平均评分】最高的 n 部电影，并提取【电影名称】和【平均评分】
    movieName_rating_top_n = movieID_rating_top_n.join(movieID_movieName).map(
        lambda x: (x[1][1], x[1][0])
    )
    return movieName_rating_top_n

Python

top10 = top(ratings, movies, 10).collect()

Python

for line in top10:
    print(line[0], ">>>rating=", line[1], sep="")

输出所有电影中最受男性喜爱的电影 Top 10 及其平均评分

Python

# 从 users 中提取出【用户 ID】和【用户性别】
userID_gender = users.map(lambda x: (int(x.split("::")[0]), x.split("::")[1]))

Python

# 将 ratings 和 userID_gender 合并后，过滤得到男性的评分数据
male_ratings = (
    ratings.map(
        lambda x: (
            int(x.split("::")[0]),
            (int(x.split("::")[1]), int(x.split("::")[2])),
        )
    )
    .join(userID_gender)
    .filter(lambda x: x[1][1] == "M")
    .map(lambda x: "{}::{}::{}".format(x[0], x[1][0][0], x[1][0][1]))
)

Python

top10_male = top(male_ratings, movies, 10).collect()

Python

for line in top10_male:
    print(line[0], ">>>rating=", line[1], sep="")

输出所有电影中最受中年人（45<=age<=55）喜爱的电影 Top 5 及其平均评分

Python

# 从 users 中提取出【用户 ID】和【年龄】
userID_age = users.map(lambda x: (int(x.split("::")[0]), int(x.split("::")[2])))

Python

# 将 ratings 和 userID_age 合并后，过滤得到中年人的评分数据
middleAge_ratings = (
    ratings.map(
        lambda x: (
            int(x.split("::")[0]),
            (int(x.split("::")[1]), int(x.split("::")[2])),
        )
    )
    .join(userID_age)
    .filter(lambda x: x[1][1] == 45 or x[1][1] == 50)
    .map(lambda x: "{}::{}::{}".format(x[0], x[1][0][0], x[1][0][1]))
)

Python

top5_middleAge = top(middleAge_ratings, movies, 5).collect()

Python

for line in top5_middleAge:
    print(line[0], ">>>rating=", line[1], sep="")

输出最受学生（含中小学生及大学生）欢迎的（即观看人数最多）动画 (Animation) 电影 Top 5 的名称

Python

# 定义函数，输入【电影 ID 和评分】、【电影 ID 和电影名称】和【观看人数前 n 位】，输出【观看人数最多的电影名称和观看次数】
def popular(ratings_data, movies_data, Genre, n):
    # 从 movies_data 中提取出【电影 ID】和【电影类型】
    movieID_movieGenre = movies_data.map(
        lambda x: (int(x.split("::")[0]), x.split("::")[2])
    )
    # 从 ratings_data 中提取出【电影 ID】，并在 value 中补充“1”作为【评分数量】，并与 movieID_movieGenre 合并，筛选得到【动画电影】的数据
    movieID_rating_number = (
        ratings_data.map(lambda x: (int(x.split("::")[1]), 1))
        .join(movieID_movieGenre)
        .filter(lambda line: Genre in line[1][1].split("|"))
        .map(lambda line: (line[0], line[1][0]))
    )
    # 对同一部电影，加总【评分数量】得到【评分数量和】
    movieID_rating_number_sum = movieID_rating_number.reduceByKey(lambda x, y: x + y)
    # 根据每一部电影的【评分数量和】从大到小排序
    movieID_rating_number_sum_descending = movieID_rating_number_sum.sortBy(
        lambda x: x[1], False
    )
    # 提取【评分数量和】最高的 n 部电影的【电影 ID】和【评分数量和】，并存入一个 RDD
    movieID_rating_number_sum_top_n = sc.parallelize(
        movieID_rating_number_sum_descending.take(n)
    )
    # 从 movies_data 中提取出【电影 ID】和【电影名称】
    movieID_movieName = movies_data.map(
        lambda x: (int(x.split("::")[0]), x.split("::")[1])
    )
    # 将【电影名称】合并到【评分数量和】最多的 n 部电影，并提取【电影名称】和【评分数量和】
    movieName_rating_number_sum_top_n = movieID_rating_number_sum_top_n.join(
        movieID_movieName
    ).map(lambda x: (x[1][1], x[1][0]))
    return movieName_rating_number_sum_top_n.sortBy(lambda x: x[1], False)

Python

# 从 users 中提取出【用户 ID】和【职业】
userID_occupation = users.map(lambda x: (int(x.split("::")[0]), int(x.split("::")[3])))

Python

# 将 ratings 和 userID_occupation 合并后，过滤得到【学生】的评分数据
student_ratings = (
    ratings.map(
        lambda x: (
            int(x.split("::")[0]),
            (int(x.split("::")[1]), int(x.split("::")[2])),
        )
    )
    .join(userID_occupation)
    .filter(lambda x: x[1][1] == 4 or x[1][1] == 10)
    .map(lambda x: "{}::{}::{}".format(x[0], x[1][0][0], x[1][0][1]))
)

Python

popular5_student_animation = popular(student_ratings, movies, "Animation", 5).collect()

Python

for line in popular5_student_animation:
    print(line[0], " >>>  ", line[1], sep="")

评论