1.1　通过RDD实战电影点评系统入门及源码阅读

日常的数据来源有很多渠道，如网络爬虫、网页埋点、系统日志等。下面的案例中使用的是用户观看电影和点评电影的行为数据，数据来源于网络上的公开数据，共有3个数据文件：user.dat、ratings.dat和movies.dat。

其中，user.dat的格式如下：

1.  UserID::Gender::Age::Occupation::Zip-code

这个文件里共有6040个用户的信息，每行中用“::”隔开的详细信息包括ID、性别（F、M分别表示女性、男性）、年龄（使用7个年龄段标记）、职业和邮编。

ratings.dat的格式如下：

1.  UserID::MovieID::Rating::Timestamp

这个文件记录的是评分信息，即用户ID、电影ID、评分（满分是5分）和时间戳。

movies.dat的格式如下：

1.  MovieID::Title::Genres

这个文件记录的是电影信息，即电影ID、电影名称和电影类型。

1.1.1　Spark核心概念图解

进入到案例实战前，首先来看几个至关重要的概念，这些概念承载着Spark集群运转和程序运行的重要使命。Spark运行架构图如图1-1所示。

Master（图1-1中的Cluster Manager）：就像Hadoop有NameNode和DataNode一样，Spark有Master和Worker。Master是集群的领导者，负责管理集群资源，接收Client提交的作业，以及向Worker发送命令。

Worker（图1-1中的Worker Node）：集群中的Worker，执行Master发送的指令，来具体分配资源，并在这些资源中执行任务。

Driver：一个Spark作业运行时会启动一个Driver进程，也是作业的主进程，负责作业的解析、生成Stage，并调度Task到Executor上。

图1-1　Spark运行架构图

Executor：真正执行作业的地方。Executor分布在集群中的Worker上，每个Executor接收Driver的命令来加载和运行Task，一个Executor可以执行一到多个Task。

SparkContext：是程序运行调度的核心，由高层调度器DAGScheduler划分程序的每个阶段，底层调度器TaskScheduler划分每个阶段的具体任务。SchedulerBackend管理整个集群中为正在运行的程序分配的计算资源Executor。

DAGScheduler：负责高层调度，划分stage并生成程序运行的有向无环图。

TaskScheduler：负责具体stage内部的底层调度，具体task的调度、容错等。

Job：（正在执行的叫ActiveJob）是Top-level的工作单位，每个Action算子都会触发一次Job，一个Job可能包含一个或多个Stage。

Stage：是用来计算中间结果的Tasksets。Tasksets中的Task逻辑对于同一个RDD内的不同partition都一样。Stage在Shuffle的地方产生，此时下一个Stage要用到上一个Stage的全部数据，所以要等到上一个Stage全部执行完才能开始。Stage有两种：ShuffleMapStage和ResultStage，除了最后一个Stage是ResultStage外，其他Stage都是ShuffleMapStage。ShuffleMapStage会产生中间结果，以文件的方式保存在集群里，Stage经常被不同的Job共享，前提是这些Job重用了同一个RDD。

Task：任务执行的工作单位，每个Task会被发送到一个节点上，每个Task对应RDD的一个partition。

RDD：是不可变的、Lazy级别的、粗粒度的（数据集级别的而不是单个数据级别的）数据集合，包含了一个或多个数据分片，即partition。

另外，Spark程序中有两种级别的算子：Transformation和Action。Transformation算子会由DAGScheduler划分到pipeline中，是Lazy级别的不会触发任务的执行；Action算子会触发Job来执行pipeline中的运算。

介绍完上面的关键概念，下面开始进入到程序编写阶段。

首先写好Spark程序的固定框架，以便于在处理和分析数据的时候专注于业务逻辑本身。

创建一个Scala的object类，在main方法中配置SparkConf和SparkContext，这里指定程序在本地运行，并且把程序名字设置为“RDD_Movie_Users_Analyzer”。

RDD_Movie_Users_Analyzer代码如下：

1.1.2　通过RDD实战电影点评系统案例

首先我们来写一个案例计算，并打印出所有电影中评分最高的前10个电影名和平均评分。

第一步：从ratingsRDD中取出MovieID和rating，从moviesRDD中取出MovieID和Name，如果后面的代码重复使用这些数据，则可以把它们缓存起来。首先把使用map算子上面的RDD中的每一个元素（即文件中的每一行）以“::”为分隔符进行拆分，然后再使用map算子从拆分后得到的数组中取出需要用到的元素，并把得到的RDD缓存起来。

第二步：从ratings的数据中使用map算子获取到形如（movieID,(rating,1)）格式的RDD，然后使用reduceByKey把每个电影的总评分以及点评人数算出来。

此时得到的RDD格式为（movieID,(Sum(ratings),Count(ratings))）。

第三步：把每个电影的Sum(ratings)和Count(ratings)相除，得到包含了电影ID和平均评分的RDD：

第四步：把avgRatings与movieInfo通过关键字（key）连接到一起，得到形如（movieID,(MovieName,AvgRating)）的RDD，然后格式化为（AvgRating,MovieName），并按照key（也就是平均评分）降序排列，最终取出前10个并打印出来。

评分最高电影运行结果如图1-2所示。

图1-2　评分最高电影运行结果

接下来我们来看另外一个功能的实现：分析最受男性喜爱的电影Top10和最受女性喜爱的电影Top10。

首先来分析一下：单从ratings中无法计算出最受男性或者女性喜爱的电影Top10，因为该RDD中没有Gender信息，如果需要使用Gender信息进行Gender的分类，此时一定需要聚合。当然，我们力求聚合使用的是mapjoin（分布式计算的一大痛点是数据倾斜，map端的join一定不会数据倾斜），这里是否可使用mapjoin？不可以，因为map端的join是使用broadcast把相对小得多的变量广播出去，这样可以减少一次shuffle，这里，用户的数据非常多，所以要使用正常的join。

使用join连接ratings和users之后，对分别过滤出男性和女性的记录进行处理：

接下来对两个RDD进行处理，处理逻辑和上面的案例相同，最终打印出来的结果分别如图1-3和图1-4所示。

图1-3　最受男性喜爱的电影运行结果

图1-4　最受女性喜爱的电影运行结果

所有电影中最受男性喜爱的电影Top10业务代码如下：

所有电影中最受女性喜爱的电影Top10业务代码如下：

在现实业务场景中，二次排序非常重要，并且经常遇到。下面来模拟一下这些场景，实现对电影评分数据进行二次排序，以Timestamp和Rating两个维度降序排列，值得一提的是，Java版本的二次排序代码非常烦琐，而使用Scala实现就会很简捷，首先我们需要一个继承自Ordered和Serializable的类。

然后再把RDD的每条记录里想要排序的字段封装到上面定义的类中作为key，把该条记录整体作为value。

取出排序后的RDD的value，此时这些记录已经是按照时间戳和评分排好序的，最终打印出的结果如图1-5所示，从图中可以看到已经按照timestamp和评分降序排列了。

图1-5　电影系统二次排序运行结果