大数据系统——Spark：Cluster Computing with Working Sets论文分享

约定不等于承诺〃 2021-12-21 10:09 325阅读 0赞

## 1.背景介绍 ##

这篇文章是UC Berkeley的AMPLab实验室发的，Berkeley也是我自己最喜欢的大学之一。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L015TnVUaA_size_16_color_FFFFFF_t_70]  
上一篇文章介绍的Pig被其作者认为是介于SQL语言和结构化程序语言之间的一种语言，而本文直接将其当作SQL接口。之前介绍的MapReduce是在磁盘框架上，而Spark是属于内存计算框架。之后还会介绍另一篇重点不同的关于Spark的论文。  
MapReduce及其变体可以在商用计算机集群上，良好运行大规模数据密集型应用程序。但是，大多数这些系统都是围绕非循环数据流模型（acyclic data flow）构建的，不适合其他流行的应用程序。本文重点介绍一种应用程序：经常重用相同的数据集，这些数据集包括多种并行操作。这包括交互性机器学习算法，以及交互式数据分析工具。提出了一个名为Spark的新框架，它支持这些应用程序，同时保留MapReduce的可扩展性和容错性。为了实现这些目标，Spark引入了一种称为弹性分布式数据集（resilient distributed datasets，RDD）。一个RDD是在一组只读集合（read-only collection），通过计算机分组，如果分组丢失，可以重建这些对象。在迭代机器学习任务中，Spark为Hadoop的10倍，并且可以以sub-second响应时间交互式查询39 GB数据集。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L015TnVUaA_size_16_color_FFFFFF_t_70 1]  
Spark通过Scala实现，Scala是Java VM的静态类型高级编程语言，并且有类似于DryadLINQ的函数编程接口。此外，通过Scala解释器的修改版本可以进行交互使用，可以用户定义RDD，函数，变量和类，在集群中并行使用它们。作者认为Spark是第一个允许有效的、通用的编程语言运行的系统，使得用户可以交互的处理数据。

## 2.Hadoop ##

Hadoop用户报告说有两种情况使用MapReduce效率不高：

1.  Iterative jobs：许多常见的机器学习算法将函数重复应用于同一数据集以优化参数（例如，通过Gradient descent）。虽然每次迭代都可以表示为一个MapReduce或Dryad作业，但每个作业都必须从磁盘重新加载数据，从而导致显着的性能损失。
2.  Interactive analytics：Hadoop通常通过诸如Pig和Hive之类的SQL接口在大型数据集上运行临时探索性查询。理想情况下，用户可以在多台计算机上将感兴趣的数据集加载到内存中并重复查询。但是，对于Hadoop，每个查询都会产生显着的延迟（几十秒），因为它作为单独的MapReduce作业运行并从磁盘读取数据。  
    Spark引入了一种重要抽象：弹性分布式数据集（RDD）。用户可以明确的在跨机器的内存中显式缓存RDD，并在多个类似MapReduce的并行操作中重复使用此RDD。RDD通过lineage这种概念来实现容错：如果RDD的分区丢失，则RDD具有关于如何从其他RDD导出，以便能够仅重建该分区的足够信息。虽然RDD不是一般的共享内存抽象，但它一方面具有较好的表达性，另一方面具有较好的扩展性和可靠性，非常适合多种应用。

## 3.编程模型 ##

使用Spark时，开发人员编写一个驱动程序（driver program）来实现高级控制流程并启动各种并行操作。Spark为并行编程提供了两个主要的抽象：弹性分布式数据集（resilient distributed datasets，RDD）和对这些数据集的并行操作（parallel operations），其中并行操作通过传递函数在数据上进行操作。此外，Spark支持两种受限类型的共享变量，可以在集群上运行的函数中使用

1.  弹性分布式数据集resilient distributed datasets，RDD：一个RDD是在一组只读集合（read-only collection），通过计算机分组，如果分组丢失，可以重建这些对象。RDD的元素不需要存在于物理存储中; 相反，RDD的句柄（handle）包含足够的信息来从可靠存储中的数据开始计算RDD。这意味着如果节点发生故障，可以始终重建RDD。在Spark中，每个RDD都由Scala对象表示。Spark允许程序员以四种方式构建RDD：（1）： 通过一个共享文件系统的文件（*file*），如HDFS；（2）通过在driver program中并行化（“*parallelizing*”）一个Scala集合（例如，一个数组），这意味着将它分成多片，进而送到各个节点；（3）通过现有RDD转化（*transforming*）。一个元素为类型A的数据集可以转化为一个元素为类型B的数据集，使用的操作为flatMap，使用用户定义的函数A=>List\[B\]来传递每一个元素；（4）通过改变现存RDD的*persistence*。默认情况下，RDD是懒惰的，也就是说，数据集的分区在并行操作中使用时（例如，通过map函数传递一些文件）按需实现，并且在使用后从内存中丢弃。然而，用户可以通过两个动作来改变RDD的persistence：(a.)cache动作让数据集lazy，但是也因此数据集在被初次计算后应该保留在内存中，因为他将会被再次用到；(b.)save动作评估数据集，并将其写到分布式文件系统，如HDFS。
2.  并行运算：RDD可以实现一些并行操作：  
    reduce：通过驱动程序将数据集元素结合起来，进而产生一个结果；  
    collect：将数据集所有元素送给驱动程序。  
    foreach：通过用户提供的函数传递每一个元素。  
    3.共享变量：程序员通过将闭包（函数）传递给Spark来调用map，filter和reduce等操作。正如典型的函数式编程中一样，这些闭包可以指向创建它们时的变量。通常，当Spark在工作节点上运行闭包时，这些变量将被复制到工作节点上。Spark还允许程序员创建两种受限类型的共享变量，以支持两种简单但常见的使用模式：  
    *Broadcast variables*：如果一块大的只读数据（例如，lookup table）被用在多个并行操作之中，则倾向于将其分发给worker仅一次，而不是将其与每一个闭包打包。Spark允许程序员创建一个Broadcast variables对象，来包装数据，并确保其只被复制到worker一次。  
    *Accumulators*：Accumulators是worker只能“add“的变量，只有driver可以读取。它们可用于实现MapReduce中的计数器，并为并行求和提供更强制的语法。任何具有”add“操作和“zero”值的类型都可以定义accumulator。因为其只读的特性，很容易实现容错机制。

## 4.代码示例 ##

1.文本搜索（text search）：假设对存储在HDFS中的日志文件中包含错误（error）的行进行计数。可以使用类似于下面的代码：

val file = spark.testFile("hdfs://...")
    	val errs = file.filter(_.contains("ERROR"))
    	val ones = errs.map(_=>1)
    	val count = ones.reduce(_+_)

首先，创建一个分布式数据集称为file，用来代表HDFS文件，形式为行的集合。然后，将此数据集转化为包含“ERROR”（errs）的行的集合，然后将每一行映射（map）为1，然后通过reduce将这些加起来。Filter，map和reduce的参数是通过Scala的语法确定的。Spark不同于其他框架的是其可以保留跨操作的中间结果。例如，如果想重新使用errs数据集，可以创建一个缓存的RDD，通过以下语句

val cachedErrs = errs.cache()

2.逻辑回归（Logistic Regression）：逻辑回归是一个迭代分类算法，算法试图找到一个hyperplane w来将点分为两部分。算法使用了梯度下降（gradient descent）：以任意值w开始，每一次迭代，加上使得w提升的值。因此，算法得益于将值缓存在内存之中，提升迭代效率。不解释逻辑回归的细节，而注重展现一些Spark新特性。

//Read points from a text file ans cache them
    	val points = spark.textFile(...).map(parsePoint).cache()
    	//Initialize w to random D-dimensional vector
    	var w = Vector.random(D)
    	//Run multiple iterations to update w
    	for(i <- 1 to ITERATIONS){
    		var grad = spark.accumulator(new Vector(D))
    		for(p <- points){//runs in parallel
    			val s = (1/(1+exp(-p.y*(w dot p.x)))-1)*p.y
    			grad += s*p.x
    		}
    		w -= grad.value
    	}

## 5.实现 ##

Spark建于Mesos之上。Mesos是集群操作系统，可以使多个并行化应用共享一个集群，同时提供API使得应用可以在集群上发布任务。这使得Spark可以与现有的集群计算框架（如Hadoop和MPI的Mesos端口）一起运行，并与它们共享数据。Spark的核心是RDD的实现。其中很重要一个概念是Lineage。例如，假设定义了一个名为cachedErrs的缓存数据集，表示日志文件中的错误消息，并且使用map和reduce计算其元素。数据集链接在一起，如图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L015TnVUaA_size_16_color_FFFFFF_t_70 2]  
每一个数据集对象包含一个指针指向其父节点，同时包含父节点如何转换为当前节点的信息。在RDD内部，每一个对象实现了一个接口，包括三个操作：getPartitions:返回一个分区ID；getIterator(partition):在分区中迭代；getPreferredLocations(partition),用来调度task，并达到数据局部性  
在数据集上调用并行操作时，Spark会创建一个task来处理数据集的每个分区，并将这些任务发送到工作节点。使用称为delay scheduling的技术将每个任务发送到其首选位置之一。不同的RDD只在其实现的接口不同。

## 6.结论 ##

RDD是Spark的一个重要抽象，使用RDD可以实现良好的容错机制：保存RDD之间的先后映射关系，出现错误时，取出父节点RDD，进而调用映射关系就可以算出出错的RDD。文章同时指出了未来的工作重点：  
1.正式表征RDD和Spark的抽象，以及对应用程序和工作负载的适用性。  
2.增强RDD的抽象，允许程序员在存储空间和重建成本之间进行自由取舍。  
3.定义新操作来转换RDD，包括通过给定键重新分配RDD的“shuffle”操作。这样的操作将允许实现group-by和join。  
4.在Spark解释器之上提供更高级别的交互式接口，例如SQL和R的shell。  
下一篇：Resilient Distributed Datasets:A Fault-Tolerant Abstraction for In-Memory Cluster Computing

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L015TnVUaA_size_16_color_FFFFFF_t_70]: /images/20211221/de97a710cb684ab18fba1b0e8b2780c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L015TnVUaA_size_16_color_FFFFFF_t_70 1]: /images/20211221/7579d534ce8a4516b632fc11a509950a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L015TnVUaA_size_16_color_FFFFFF_t_70 2]: /images/20211221/dabfb283f0594e5c88ce41f656539aeb.png