与许多专有的大数据处理平台不同，Spark建立在统一抽象的RDD之上，使得它可以以基本一致的方式应对不同的大数据处理场景，包括MapReduce，Streaming，SQL，Machine Learning以及Graph等。这即Matei Zaharia所谓的“设计一个通用的编程抽象（Unified Programming Abstraction）。这正是Spark这朵小火花让人着迷的地方。
要理解Spark，就需得理解RDD。
RDD是什么？
RDD，全称为Resilient Distributed Datasets，是一个容错的、并行的数据结构，可以让用户显式地将数据存储到磁盘和内存中，并能控制数据的分区。同时，RDD还提供了一组丰富的操作来操作这些数据。在这些操作中，诸如map、flatMap、filter等转换操作实现了monad模式，很好地契合了Scala的集合操作。除此之外，RDD还提供了诸如join、groupBy、reduceByKey等更为方便的操作（注意，reduceByKey是action，而非transformation），以支持常见的数据运算。
通常来讲，针对数据处理有几种常见模型，包括：Iterative Algorithms，Relational Queries，MapReduce，Stream Processing。例如Hadoop MapReduce采用了MapReduces模型，Storm则采用了Stream Processing模型。RDD混合了这四种模型，使得Spark可以应用于各种大数据处理场景。
RDD（Resilient Distributed Datasets弹性分布式数据集），是spark中最重要的概念，可以简单的把RDD理解成一个提供了许多操作接口的数据集合，和一般数据集不同的是，其实际数据分布存储于一批机器中（内存或磁盘中）。当然，RDD肯定不会这么简单，它的功能还包括容错、集合内的数据可以并行处理等。图1是RDD类的视图。
图1
1、RDD是什么 RDD：Spark的核心概念是RDD (resilientdistributed dataset)，指的是一个只读的，可分区的分布式数据集，这个数据集的全部或部分可以缓存在内存中，在多次计算间重用。 为什么会产生RDD？ （1）传统的MapReduce虽然具有自动容错、平衡负载和可拓展性的优点，但是其最大缺点是采用非循环式的数据流模型，使得在迭代计算式要进行大量的磁盘IO操作。RDD正是解决这一缺点的抽象方法 （2）RDD的具体描述RDD（弹性数据集）是Spark提供的最重要的抽象的概念，它是一种有容错机制的特殊集合，可以分布在集群的节点上，以函数式编操作集合的方式，进行各种并行操作。可以将RDD理解为一个具有容错机制的特殊集合，它提供了一种只读、只能有已存在的RDD变换而来的共享内存，然后将所有数据都加载到内存中，方便进行多次重用。a.他是分布式的，可以分布在多台机器上，进行计算。b.他是弹性的，计算过程中内错不够时它会和磁盘进行数据交换。c.这些限制可以极大的降低自动容错开销d.实质是一种更为通用的迭代并行计算框架，用户可以显示的控制计算的中间结果，然后将其自由运用于之后的计算。 （3）RDD的容错机制实现分布式数据集容错方法有两种：数据检查点和记录更新RDD采用记录更新的方式：记录所有更新点的成本很高。所以，RDD只支持粗颗粒变换，即只记录单个块上执行的单个操作，然后创建某个RDD的变换序列（血统）存储下来；变换序列指，每个RDD都包含了他是如何由其他RDD变换过来的以及如何重建某一块数据的信息。因此RDD的容错机制又称“血统”容错。 要实现这种“血统”容错机制，最大的难题就是如何表达父RDD和子RDD之间的依赖关系。实际上依赖关系可以分两种，窄依赖和宽依赖：窄依赖：子RDD中的每个数据块只依赖于父RDD中对应的有限个固定的数据块；宽依赖：子RDD中的一个数据块可以依赖于父RDD中的所有数据块。例如：map变换，子RDD中的数据块只依赖于父RDD中对应的一个数据块；groupByKey变换，子RDD中的数据块会依赖于多有父RDD中的数据块，因为一个key可能错在于父RDD的任何一个数据块中 将依赖关系分类的两个特性：第一，窄依赖可以在某个计算节点上直接通过计算父RDD的某块数据计算得到子RDD对应的某块数据；宽依赖则要等到父RDD所有数据都计算完成之后，并且父RDD的计算结果进行hash并传到对应节点上之后才能计算子RDD。第二，数据丢失时，对于窄依赖只需要重新计算丢失的那一块数据来恢复；对于宽依赖则要将祖先RDD中的所有数据块全部重新计算来恢复。所以在长“血统”链特别是有宽依赖的时候，需要在适当的时机设置数据检查点。也是这两个特性要求对于不同依赖关系要采取不同的任务调度机制和容错恢复机制。 （4）RDD内部的设计每个RDD都需要包含以下四个部分：a.源数据分割后的数据块，源代码中的splits变量b.关于“血统”的信息，源码中的dependencies变量c.一个计算函数（该RDD如何通过父RDD计算得到），源码中的iterator(split)和compute函数d.一些关于如何分块和数据存放位置的元信息，如源码中的partitioner和preferredLocations例如：a.一个从分布式文件系统中的文件得到的RDD具有的数据块通过切分各个文件得到的，它是没有父RDD的，它的计算函数知识读取文件的每一行并作为一个元素返回给RDD；b.对与一个通过map函数得到的RDD，它会具有和父RDD相同的数据块，它的计算函数式对每个父RDD中的元素所执行的一个函数 2、RDD在Spark中的地位及作用 （1）为什么会有Spark？因为传统的并行计算模型无法有效的解决迭代计算（iterative）和交互式计算（interactive）；而Spark的使命便是解决这两个问题，这也是他存在的价值和理由。 （2）Spark如何解决迭代计算？其主要实现思想就是RDD，把所有计算的数据保存在分布式的内存中。迭代计算通常情况下都是对同一个数据集做反复的迭代计算，数据在内存中将大大提升IO操作。这也是Spark涉及的核心：内存计算。 （3）Spark如何实现交互式计算？因为Spark是用scala语言实现的，Spark和scala能够紧密的集成，所以Spark可以完美的运用scala的解释器，使得其中的scala可以向操作本地集合对象一样轻松操作分布式数据集。 （4）Spark和RDD的关系？可以理解为：RDD是一种具有容错性基于内存的集群计算抽象方法，Spark则是这个抽象方法的实现。 3、如何操作RDD？ （1）如何获取RDDa.从共享的文件系统获取，（如：HDFS）b.通过已存在的RDD转换c.将已存在scala集合（只要是Seq对象）并行化 ，通过调用SparkContext的parallelize方法实现d.改变现有RDD的之久性；RDD是懒散，短暂的。（RDD的固化：cache缓存至内错； save保存到分布式文件系统） （2）操作RDD的两个动作a.Actions：对数据集计算后返回一个数值value给驱动程序；例如：Reduce将数据集的所有元素用某个函数聚合后，将最终结果返回给程序。b.Transformation：根据数据集创建一个新的数据集，计算后返回一个新RDD；例如：Map将数据的每个元素经过某个函数计算后，返回一个姓的分布式数据集。 （3）Actions具体内容：
reduce(func)通过函数func聚集数据集中的所有元素。Func函数接受2个参数，返回一个值。这个函数必须是关联性的，确保可以被正确的并发执行collect()在Driver的程序中，以数组的形式，返回数据集的所有元素。这通常会在使用filter或者其它操作后，返回一个足够小的数据子集再使用，直接将整个RDD集Collect返回，很可能会让Driver程序OOMcount()返回数据集的元素个数take(n)返回一个数组，由数据集的前n个元素组成。注意，这个操作目前并非在多个节点上，并行执行，而是Driver程序所在机器，单机计算所有的元素(Gateway的内存压力会增大，需要谨慎使用）first()返回数据集的第一个元素（类似于take（1）saveAsTextFile(path)将数据集的元素，以textfile的形式，保存到本地文件系统，hdfs或者任何其它hadoop支持的文件系统。Spark将会调用每个元素的toString方法，并将它转换为文件中的一行文本saveAsSequenceFile(path)将数据集的元素，以sequencefile的格式，保存到指定的目录下，本地系统，hdfs或者任何其它hadoop支持的文件系统。RDD的元素必须由key-value对组成，并都实现了Hadoop的Writable接口，或隐式可以转换为Writable（Spark包括了基本类型的转换，例如Int，Double，String等等）foreach(func)在数据集的每一个元素上，运行函数func。这通常用于更新一个累加器变量，或者和外部存储系统做交互 （4）Transformation具体内容
map(func)
返回一个新的分布式数据集，由每个原元素经过func函数转换后组成
filter(func)
返回一个新的数据集，由经过func函数后返回值为true的原元素组成
flatMap(func)
类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）
flatMap(func)
类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）
sample(withReplacement, frac, seed)
根据给定的随机种子seed，随机抽样出数量为frac的数据
union(otherDataset)
返回一个新的数据集，由原数据集和参数联合而成
groupByKey([numTasks])
在一个由（K,V）对组成的数据集上调用，返回一个（K，Seq[V])对的数据集。注意：默认情况下，使用8个并行任务进行分组，你可以传入numTask可选参数，根据数据量设置不同数目的Task
reduceByKey(func, [numTasks])
在一个（K，V)对的数据集上使用，返回一个（K，V）对的数据集，key相同的值，都被使用指定的reduce函数聚合到一起。和groupbykey类似，任务的个数是可以通过第二个可选参数来配置的。
join(otherDataset, [numTasks])
在类型为（K,V)和（K,W)类型的数据集上调用，返回一个（K,(V,W))对，每个key中的所有元素都在一起的数据集
groupWith(otherDataset, [numTasks])
在类型为（K,V)和(K,W)类型的数据集上调用，返回一个数据集，组成元素为（K, Seq[V], Seq[W]) Tuples。这个操作在其它框架，称为CoGroup
cartesian(otherDataset)
笛卡尔积。但在数据集T和U上调用时，返回一个(T，U）对的数据集，所有元素交互进行笛卡尔积。
flatMap(func)
类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）
4、 一个简单的例子
下面是一个实用scala语言编写的spark应用（摘自Apache Spark 社区https://spark.apache.org/docs/latest/quick-start.html）。
/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
object SimpleApp {
def main(args: Array[String]) {
val logFile = "YOUR_SPARK_HOME/README.md" // Should be some file on your system
val conf = new SparkConf().setAppName("Simple Application") //设置程序名字
val sc = new SparkContext(conf)
val logData = sc.textFile(logFile, 2).cache() //加载文件为RDD，并缓存
val numAs = logData.filter(line => line.contains("a")).count()//包含a的行数
val numBs = logData.filter(line => line.contains("b")).count()//包含b的行数
println("Lines with a: %s, Lines with b: %s".format(numAs, numBs))
}
}
这个程序只是简单的对输入文件README.md包含'a'和'b'的行分别计数。当然如果你想运行这个程序，需要把YOUR_SPARK_HOME替换为Spark的安装目录。程序中定义了一个RDD：logData，并调用cache，把RDD数据缓存在内存中，这样能防止重复加载文件。filter是RDD提供的一种操作，它能过滤出符合条件的数据，count是RDD提供的另一个操作，它能返回RDD数据集中的记录条数。
5、RDD操作类型
上述例子介绍了两种RDD的操作：filter与count；事实上，RDD还提供了许多操作方法，如map，groupByKey，reduce等操作。RDD的操作类型分为两类，转换（transformations），它将根据原有的RDD创建一个新的RDD；行动（actions），对RDD操作后把结果返回给driver。例如，map是一个转换，它把数据集中的每个元素经过一个方法处理后返回一个新的RDD；而reduce则是一个action，它收集RDD的所有数据后经过一些方法的处理，最后把结果返回给driver。
RDD的所有转换操作都是lazy模式，即Spark不会立刻计算结果，而只是简单的记住所有对数据集的转换操作。这些转换只有遇到action操作的时候才会开始计算。这样的设计使得Spark更加的高效，例如，对一个输入数据做一次map操作后进行reduce操作，只有reduce的结果返回给driver，而不是把数据量更大的map操作后的数据集传递给driver。
RDD底层实现原理
RDD是一个分布式数据集，顾名思义，其数据应该分部存储于多台机器上。事实上，每个RDD的数据都以Block的形式存储于多台机器上，下图是Spark的RDD存储架构图，其中每个Executor会启动一个BlockManagerSlave，并管理一部分Block；而Block的元数据由Driver节点的BlockManagerMaster保存。BlockManagerSlave生成Block后向BlockManagerMaster注册该Block，BlockManagerMaster管理RDD与Block的关系，当RDD不再需要存储的时候，将向BlockManagerSlave发送指令删除相应的Block。
图2 RDD存储原理
RDD cache的原理
RDD的转换过程中，并不是每个RDD都会存储，如果某个RDD会被重复使用，或者计算其代价很高，那么可以通过显示调用RDD提供的cache()方法，把该RDD存储下来。那RDD的cache是如何实现的呢？
RDD中提供的cache()方法只是简单的把该RDD放到cache列表中。当RDD的iterator被调用时，通过CacheManager把RDD计算出来，并存储到BlockManager中，下次获取该RDD的数据时便可直接通过CacheManager从BlockManager读出。
RDD dependency与DAG
RDD提供了许多转换操作，每个转换操作都会生成新的RDD，这是新的RDD便依赖于原有的RDD，这种RDD之间的依赖关系最终形成了DAG（Directed Acyclic Graph）。
RDD之间的依赖关系分为两种，分别是NarrowDependency与ShuffleDependency，其中ShuffleDependency为子RDD的每个Partition都依赖于父RDD的所有Partition，而NarrowDependency则只依赖一个或部分的Partition。下图的groupBy与join操作是ShuffleDependency，map和union是NarrowDependency。
RDD作为数据结构，本质上是一个只读的分区记录集合。一个RDD可以包含多个分区，每个分区就是一个dataset片段。RDD可以相互依赖。如果RDD的每个分区最多只能被一个Child RDD的一个分区使用，则称之为narrow dependency；若多个Child RDD分区都可以依赖，则称之为wide dependency。不同的操作依据其特性，可能会产生不同的依赖。例如map操作会产生narrow dependency，而join操作则产生wide dependency。
Spark之所以将依赖分为narrow与wide，基于两点原因。
首先，narrow dependencies可以支持在同一个cluster node上以管道形式执行多条命令，例如在执行了map后，紧接着执行filter。相反，wide dependencies需要所有的父分区都是可用的，可能还需要调用类似MapReduce之类的操作进行跨节点传递。
其次，则是从失败恢复的角度考虑。narrow dependencies的失败恢复更有效，因为它只需要重新计算丢失的parent partition即可，而且可以并行地在不同节点进行重计算。而wide dependencies牵涉到RDD各级的多个Parent Partitions。
图3 RDD dependency
RDD partitioner与并行度
每个RDD都有Partitioner属性，它决定了该RDD如何分区，当然Partition的个数还将决定每个Stage的Task个数。当前Spark需要应用设置Stage的并行Task个数（配置项为：spark.default.parallelism），在未设置的情况下，子RDD会根据父RDD的Partition决定，如map操作下子RDD的Partition与父Partition完全一致，Union操作时子RDD的Partition个数为父Partition个数之和。
如何设置spark.default.parallelism对用户是一个挑战，它会很大程度上决定Spark程序的性能。
摘自：http://shiyanjun.cn/archives/744.html
http://www.aboutyun.com/thread-7214-1-1.html
转载请注明原文地址: https://ju.6miu.com/read-11997.html
