本文共 5836 字，大约阅读时间需要 19 分钟。

RDD产生原因

MapReduce计算结果需要保存到磁盘上，Spark把中间数据放到内存中，效率高，且支持迭代计算

RDD概念 

 弹性分布式数据集

RDD属性

1、分区（Partition）：数据集的基本组成单位。每个分区都会被一个计算任务处理，并决定并行计算的粒度。用户可在创建RDD时指定RDD的分区个数，如没有指定，会采用默认值。默认值就是程序所分配到的CPU Core的数目

 

2、依赖关系：RDD每次转换都会生成一个新RDD，RDD之间会形成类似流水线一样的前后依赖关系。在部分分区数据丢失时，Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算

3、分区函数：当前Spark中实现了两种类型的分区函数，一个是基于哈希的HashPartitioner，另外一个是基于范围的RangePartitioner。只有对于key-value的RDD，才会有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函数不但决定了RDD本身的分区数量，也决定了parent RDD Shuffle输出时的分区数量。

4、列表：存储每个Partition所在的块的位置。按照移动数据不如移动计算的理念，Spark在任务调度时，会尽可能地将计算任务分配到其所要处理数据块的存储位置

RDD的弹性

1、自动进行内存和磁盘数据存储切换

Spark优先把数据放到内存中，如果内存放不下，才放到磁盘里面，程序自动进行存储切换

2、基于血统的高效容错机制

在RDD进行转换和动作的时候，会形成RDD的Lineage依赖链，当某一个RDD失效的时候，可以通过重新计算上游的RDD来重新生成丢失RDD数据

3、Task如果失败会自动进行特定次数的重试

RDD的计算任务如果运行失败，会自动进行任务的重新计算，默认次数是4次

RDD特点

1、分区：RDD是分区的，计算时通过compute函数得到每个分区数据

 

2、只读：想改变RDD中的数据，只能在现有的RDD基础上创建新的RDD

3、依赖：依赖包括两种，一种是窄依赖，另一种是宽依赖

4、缓存：应用程序中多次使用同一个RDD，会将该RDD缓存起来，该RDD只有在第一次计算的时候会根据血缘关系得到分区的数据，在后续其他地方用到该RDD的时候，会直接从缓存处取而不用再根据血缘关系计算

编程模型

编写Driver程序被提交到集群(Application)以调度运行Worker。下图Driver中定义了一个或多个RDD，并调用RDD上的action，Worker执行RDD分区计算任务

RDD创建

在Spark中创建RDD的创建方式可分为三种：

1、从集合中创建：由一个已经存在的Scala集合创建，集合并行化

2、从外部存储创建：读取文件生成

3、从其他RDD创建：RDD可以通过其他的RDD转换而来的

TransFormation

RDD编程APISpark支持两个类型（算子）操作：Transformation和Action

Transformation就是将一个已有的RDD生成另一个RDD。Transformation具有lazy特性(延迟加载)。Transformation算子代码不会真正被执行，只有当程序中遇到一个action算子时，代码才会真正被执行

转换	含义
map(func)	返回一个新的RDD，该RDD由每一个输入元素经过func函数转换后组成
filter(func)	返回一个新的RDD，该RDD由经过func函数计算后返回值为true的输入元素组成
flatMap(func)	类似于map，但是每一个输入元素可以被映射为0或多个输出元素（所以func应该返回一个序列，而不是单一元素）
mapPartitions(func)	类似于map，但独立地在RDD的每一个分区上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func)	类似于mapPartitions，但func带有一个整数参数表示分区的索引值，因此在类型为T的RDD上运行时，func的函数类型必须是 (Int, Interator[T]) => Iterator[U]
union(otherDataset)	对源RDD和参数RDD求并集后返回一个新的RDD
intersection(otherDataset)	对源RDD和参数RDD求交集后返回一个新的RDD（交集）
distinct([numTasks]))	对源RDD进行去重后返回一个新的RDD
groupByKey([numTasks])	在一个(K,V)的RDD上调用，返回一个(K, Iterator[V])的RDD
reduceByKey(func, [numTasks])	在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，与groupByKey类似，reduce任务的个数可以通过第二个可选的参数来设置
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks])	先按分区聚合 再总的聚合   每次要跟初始值交流 例如：aggregateByKey(0)(_+_,_+_) 对k/y的RDD进行操作
sortByKey([ascending], [numTasks])	在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD
sortBy(func,[ascending], [numTasks])	与sortByKey类似，但是更灵活 第一个参数是根据什么排序  第二个是怎么排序 false倒序   第三个排序后分区数  默认与原RDD一样
join(otherDataset, [numTasks])	在类型为(K,V)和(K,W)的RDD上调用，返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD  相当于内连接（求交集）
cogroup(otherDataset, [numTasks])	在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD
cartesian(otherDataset)	两个RDD的笛卡尔积  的成很多个K/V
coalesce(numPartitions)	重新分区 第一个参数是要分多少区，第二个参数是否shuffle 默认false  少分区变多分区 true   多分区变少分区 false
repartition(numPartitions)	重新分区 必须shuffle  参数是要分多少区  少变多
foldByKey(zeroValue)(seqOp)	该函数用于K/V做折叠，合并处理 ，与aggregate类似   第一个括号的参数应用于每个V值  第二括号函数是聚合例如：_+_
combineByKey	合并相同的key的值 rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n)
partitionBy（partitioner）	对RDD进行分区  partitioner是分区器 例如new HashPartition(2
cache	RDD缓存，可以避免重复计算从而减少时间，区别：cache内部调用了persist算子，cache默认就一个缓存级别MEMORY-ONLY ，而persist则可以选择缓存级别
persist
Subtract（rdd）	返回前rdd元素不在后rdd的rdd（差集）
leftOuterJoin	leftOuterJoin类似于SQL中的左外关联left outer join，返回结果以前面的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可。
rightOuterJoin	rightOuterJoin类似于SQL中的有外关联right outer join，返回结果以参数中的RDD为主，关联不上的记录为空。只能用于两个RDD之间的关联，如果要多个RDD关联，多关联几次即可
subtractByKey	substractByKey和基本转换操作中的subtract类似只不过这里是针对K的，返回在主RDD中出现，并且不在otherRDD中出现的元素

Action

触发代码的运行，一段spark代码里至少需要一个action操作

动作	含义
reduce(func)	通过func函数聚集RDD中的所有元素，这个功能必须是课交换且可并联的
collect()	在驱动程序中，以数组的形式返回数据集的所有元素
count()	返回RDD的元素个数
first()	返回RDD的第一个元素（类似于take(1)）
take(n)	返回一个由数据集的前n个元素组成的数组
takeSample(withReplacement,num, [seed])	返回一个数组，该数组由从数据集中随机采样的num个元素组成，可以选择是否用随机数替换不足的部分，seed用于指定随机数生成器种子
takeOrdered(n, [ordering])	使用自然顺序或自定义比较器返回RDD 的前n个元素。
saveAsTextFile(path)	将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统，对于每个元素，Spark将会调用toString方法，将它装换为文件中的文本
saveAsSequenceFile(path)	将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下，可以使HDFS或者其他Hadoop支持的文件系统。
saveAsObjectFile(path)	使用Java序列化以简单格式编写数据集的元素，然后可以使用Java序列化加载SparkContext.objectFile()。
countByKey()	针对(K,V)类型的RDD，返回一个(K,Int)的map，表示每一个key对应的元素个数。
foreach(func)	在数据集的每一个元素上，运行函数func进行更新。

广播变量

1、广播变量的意义

如果在分布式计算里分发集合对象，由Driver端分发，每个task就会分发一份。在task多的情况下，会大量消耗task服务器上的资源。如果将这个变量声明为广播变量，那么只是每个executor拥有一份，task会共享这个变量，节省服务器的资源

2、图解

不使用广播变量

 

使用广播变量的情况

 

3、定义广播变量

val a = 3

val broadcast = sc.broadcast(a)

4、还原广播变量

val c = broadcast.value

5、广播变量使用

使用广播变量的过程如下：

1、调用 SparkContext.broadcast 创建出一个 Broadcast[T]对象，任何可序列化的类型都可以这么实现

2、通过 value 属性访问该对象的值(在Java中为value()方法)

3、变量只会被发到各个节点一次，应作为只读值处理(修改这个值不会影响到别的节点)

val conf = new SparkConf()conf.setMaster("local").setAppName("brocast")val sc = new SparkContext(conf)val list = List("hello hadoop")val broadCast = sc.broadcast(list)val lineRDD = sc.textFile("./words.txt")lineRDD.filter { x => broadCast.value.contains(x) }.foreach { println}sc.stop()

6、定义广播变量注意点

广播变量只读，不能修改

7、注意事项

1、能不能将一个RDD使用广播变量广播出去？

      不能，因为RDD是不存储数据的。可以将RDD的结果广播出去

2、 广播变量只能在Driver端定义，不能在Executor端定义

3、 在Driver端可以修改广播变量的值，在Executor端无法修改广播变量的值

4、如果executor端用到了Driver的变量，如果不使用广播变量在Executor有多少task就有多少Driver端的变量副本

5、如果Executor端用到了Driver的变量，如果使用广播变量在每个Executor中只有一份Driver端的变量副本

累加器

1、累加器的意义

Spark在分布式运行时每个task运行的是原始变量的一个副本，不能改变原始变量的值，当这个变量被声明为累加器后，该变量就会有分布式计数的功能

2、图解累加器

不使用累加器

 

使用累加器

 

3、定义累加器

val a = sc.accumulator(0)

4、还原累加器

val b = a.value

5、累加器的使用

val conf = new SparkConf()conf.setMaster("local").setAppName("accumulator")val sc = new SparkContext(conf)val accumulator = sc.accumulator(0)sc.textFile("./words.txt").foreach { x =>{accumulator.add(1)}}println(accumulator.value)sc.stop()

驱动器中调用SparkContext.accumulator(initialValue)方法，创建出存有初始值的累加器。使用累加器的 add方法增加累加器的值

6、注意：工作节点上的任务不能访问累加器的值，累加器是一个只写变量

转载地址：http://ovazi.baihongyu.com/