Spark:Cluster Computing with Working Sets 论文阅读笔记

本文由UCB发表于2010年，10年也是MR还处盛行的时候，因此文中会频繁将Spark与MR对比。关注两个框架对比的也可以看博主的另一篇博文：Spark学习笔记之（一）：MR与Spark的区别：
https://blog.csdn.net/u010737756/article/details/118406700?spm=1001.2014.3001.5501
从计算速度、资源、容错、功能适用、生态、运行环境六个角度简述了二者的区别。

回到论文，该文章介绍了基于RDD的分布式计算模型以及早期Spark的实现。文中通过与MR对比，提到了Spark与MR类似，提供了较强的可扩展性与高容错。此外，Spark提出框架优于MR的场景，例如：在同一数据处于反复迭代计算的场景下，MR需要重复从磁盘读取数据，而Spark可通过多级别的持久化机制，将数据存在内存或分布式文件系统中，进行RDD的重用等等。

这篇论文阅读笔记会从文中描述的Spark适用场景例子入手，简要介绍一下2010年设计之初的Spark的设计目的，以及这十年来，令Spark一直保持分布式计算框架核心竞争力的部分特性。

在Abstract中，作者提出Spark这个新框架是针对数据集在并行计算操作场景下，更适用于需要迭代计算的机器学习场景，以及交互式查询、分析场景这两种使用场景。于此同时，Spark框架也保持了MR的可扩展性与高容错。

而Spark在这两个用途的高性能、可扩展性、高容错，也正是Spark贯穿本文的，设计之初的最大目的。

文中举例了两种，一种是逻辑回归、一种是交替最小二乘法。

背景：给定一组点集，通过迭代分类算法，试图找到一个超平面w把两个点集合分隔开。
该算法采用梯度下降法，开始给w赋一个随机值。
在每次迭代中对w的结果进行修正，移动w的方向对结果进行优化。
首先创建一个名为points的RDD节点，我们通过运行一个循环来处理它。
for关键字是Scala里面用于表示调用循环的语法，里面的循环体类似于foreach方法。
代码"for(p <- points){body}“等同于"points.foreach(p =>{body})”，在此调用了Spark的并行foreach操作。
其次定义了一个名为grad的梯度累加器（类型为Vector）。需要注意的是，循环体中的累加是并行执行的。

处理29GB的数据。采用20个"m1.xlarge” EC2节点，每个节点有4个处理器核，进行逻辑回归计算。
Hadoop vs. Spark：
Hadoop每次迭代任务耗时127秒，每个MapReduce任务都是独立运行。
Spark第一次迭代用了174秒（文中解释可能由于使用Scala代替了Java），后续迭代都只需要6秒，因为每个缓存中的数据多可以复用，这使得运行速度加快了10倍以上。

值得注意的是，在部分不需要迭代计算的部分场景下，如文中提到逻辑回归的首次迭代计算中，MR是快于Spark的。也就是说，Spark虽然是个新框架，但并不会在所有使用场景下都优于MR，尤其是不需要迭代计算的场景下。

背景：交替最小二乘法用于处理协同过滤的问题，例如要通过用户对电影观看历史和评分来预测他们喜欢的电影。该算法是CPU密集型的，而不是数据密集型的。
假设我们需要预测用户u对电影m的评分，而我们已经有了很多以往用户对电影的观看数据矩阵R。
模型R是两个矩阵M和U的运算结果，M和U的尺寸分别是 M * U 和 K * U。
每个用户和影片都有一个K维的“特征向量”，描述了它的特点和用户给予它的评价。
该特征向量就是用户评级和电影的特点的内积。
交替最小二乘法解决了使用已知的观看评价的M和U，然后计算M*U矩阵的未知值的预测算法。以下使用迭代过程来实现：
1、使用随机值初始化M
2、计算优化U给定M的预测模型R，最大限度的减少错误。
3、计算优化M给定U的预测模型R，最大限度的减少错误。
4、重复2、3两步，直到收敛。

如论文第四页的Implementation章节所述，在内部，每一个RDD对象都实现了三个简要接口，包括如下三个操作：
getPartitions：返回数据分块ID的列表。
getIterator(partition)：迭代一个数据分块
getPreferredLocations(partition)：用来进行任务调度，以实现数据局部特性。

当数据集被调用进行并行操作时Spark创建一个任务并将任务分发到节点处理。而Spark会设法把每个任务都发送到其首选的位置（最优位置），这种技术称之为“延迟调度”（delay scheduling）。一旦worker开始工作，那么处理任务时都需要用getIterator方法来对数据分块进行读取。
不同类型的RDD之间只是接口不同。例如对于一个HdfsTextFile,该数据分块就是HDFS块上的ID，首选的位置就是block的位置。getIterator打开一个数据流用以读取block。
将作业传递给workers这一过程需要通过发送闭包给workers完成。闭包既可以用来定义一个分布式数据集，也可以用来传递reduce这样的操作。
Scala闭包也是Java对象，也可以通过Java序列化机制进行序列化。这是Scala的特性，可以相对简单地将各种计算处理的过程直接发送到另外一台机器上。
最后，共享变量broadcast，会使用带有序列化格式的自定义类来实现。

所以在论文第五页的Interpreter Integration章节，Spark整合了Scala的解释器。并做了两点变化：
1、实现了一个共享文件系统的解释器输出类，使得worker能通过自定义的java类加载器加载它们。输出类到共享文件系统。
2、修改了代码生成逻辑，使得每行上创建的单例对象能直接引用各个行对象的实例，而不是通过静态方法getInstance（如果通过静态方法，那对象是传不到Workers的）。

测试条件：5000部电影以及15000个用户的数据量，在30个节点的EC2集群工作集上运行。
Hadoop vs. Spark：
Spark相比Hadoop性能提升了2.8倍。

针对现有数据库查询计算是决策支持的基础，交互式查询是终端用户的最基本需求，准确完备的检索条件可以更好地帮助用户从数据库获取最需要的信息。
交互式查询的解决方案主要有两种：一种是实现交互式查询运算的工具，最通用的就是通过SQL语句，直接由数据库查询。第二种是进行交互式查询运算，也可以通过直接编写程序来实现。但是和使用SQL相比难度更高，工作量更大。同时，程序编写完成后不易修改，如果业务逻辑发生变化，或者临时遇到新的计算任务，都不能及时处理。

论文的第六页的Cluster Computing frameworks章节提到，当时有论文中提出了MR的局限性，以及提出了扩展MR提高可实现性的可能性。
论文参考：Twister: 迭代MapReduce计算框架—— https://blog.csdn.net/baidu_35570545/article/details/73275279
但是Twister作为一个没有底层分布式文件系统，纯靠网络信息传递来实现所用通讯和数据传输的计算框架，虽然提高了计算性能，但是并没有实现高容错。而Spark做到了，Spark通过定义多个能够在算子计算过程中，可进行多种选择性持久化的RDDs，以及提供多个可选择性的操作算子，来实现了迭代计算与交互性查询的优秀性能。

在论文第一页的Introduction部分的右侧段落，作者就迫不及待提出了Spark的高容错优化。

Spark通过弹性分布式数据集RDD，一个分布式的只读对象合集。文中提到，如果RDD的部分分区丢失，用户可以将RDD持久化在内存中，可以并行着进行RDD的重用。Lineage血缘机制被引入Spark，如果一个RDD的某个分区丢失，该RDD完全有充分的信息如何将这部分分区中的数据从其他RDD中重建出来。(当时没有提DAG)

正如文中第二页在介绍RDD的结尾处所说：
Our goal is to let users trade off between 1)the cost of storing an RDD, 2)the speed of accessing it, 3)the probability of losing part of it, 4)and the cost of recomputing it.
设计目标是需要找到：读、写性能；容错、错误重建性能的最优解。

论文的第六页的7.Discussion and future work章节提到作者未来对Spark的几点优化方向，而现在在2021年翻看这篇论文，也能看到后续在文中提到的以下四点，作者都进行了优化与改进。

1. Formally characterize the properties of RDDs and Spark’s other abstractions, and their suitability for various classes of applications and workloads. 面对Spark的优化特性，Spark进行了进行不同种类应用与负载的探索。包括后续的GraphX、MLlib
2. Enhance the RDD abstraction to allow programmers to trade between storage cost and re-construction cost. 优化RDD，平衡存储与重建的性能
3. Define new operations to transform RDDs, including a “shuffle” operation that repartitions an RDD by a
   given key. Such an operation would allow us to implement group-bys and joins.
   创建更多功能特性的算子
4. Provide higher-level interactive interfaces on top of the Spark interpreter, such as SQL and R shells
   提供高级的交互性查询API，如现在的Spark SQL。
   以上四点外，Spark后续也推出了Spark Streaming微批计算框架，进行准实时计算。

当初论文中只跑在UCB自行研发的Mesos上的Spark，在十多年后的如今，已经广泛应用于诸多行业的生产运营。