Spark框架核心概念

时间 2019-11-13

标签 spark 框架核心概念栏目 Spark 繁體版

原文原文链接

Spark框架核心概念

首先介绍Spark中的核心名词概念，而后再逐一详细说明。html

RDD：弹性分布式数据集，是Spark最核心的数据结构。有分区机制，因此能够分布式进行处理。有容错机制，经过RDD之间的依赖关系来恢复数据。java

依赖关系：RDD的依赖关系是经过各类Transformation（变换）来获得的。父RDD和子RDD之间的依赖关系分两种：①窄依赖②宽依赖。web

①窄依赖：父RDD的分区和子RDD的分区关系是：一对一。redis

窄依赖不会发生Shuffle，执行效率高，spark框架底层会针对多个连续的窄依赖执行流水线优化，从而提升性能。例如map、flatMap等方法都是窄依赖方法。算法

②宽依赖：父RDD的分区和子RDD的分区关系是：一对多。apache

宽依赖会产生shuffle，会产生磁盘读写，没法优化。编程

DAG：有向无环图，当一个RDD的依赖关系造成以后，就造成了一个DAG。通常来讲，一个DAG，最后都至少会触发一个Action操做，触发执行。一个Action对应一个Job任务。api

Stage：一个DAG会根据RDD之间的依赖关系进行Stage划分，流程是：以Action为基准，向前回溯，遇到宽依赖，就造成一个Stage。遇到窄依赖，则执行流水线优化（将多个连续的窄依赖放到一块儿执行）。数组

task：任务。一个分区对应一个task。能够这样理解：一个Stage是一组Task的集合。缓存

RDD的Transformation（变换）操做：懒执行，并不会当即执行。

RDD的Action(执行）操做：触发真正的执行。

一、RDD

Resilient Distributed Datasets (RDDs)

Spark revolves around the concept of a resilient distributed dataset (RDD), which is a fault-tolerant collection of elements that can be operated on in parallel. There are two ways to create RDDs: parallelizing an existing collection in your driver program, or referencing a dataset in an external storage system, such as a shared filesystem, HDFS, HBase, or any data source offering a Hadoop InputFormat.

RDD弹性分布式数据集：就是带有分区的集合类型。特色是能够并行操做，而且是容错的。

有两种方法能够建立RDD：

1）执行Transform操做（变换操做），

2）读取外部存储系统的数据集，如HDFS，HBase，或任何与Hadoop有关的数据源。

1．RDD入门示例

案例一

Parallelized collections are created by calling SparkContext’s parallelize method on an existing collection in your driver program (a Scala Seq). The elements of the collection are copied to form a distributed dataset that can be operated on in parallel. For example, here is how to create a parallelized collection holding the numbers 1 to 5:

val data = Array(1, 2, 3, 4, 5)
val r1 = sc.parallelize(data)
val r2 = sc.parallelize(data,2)

你能够这样理解RDD：

它是spark提供的一个特殊集合类。诸如普通的集合类型，如传统的Array：（1,2,3,4,5）是一个总体，但转换成RDD后，咱们能够对数据进行Partition（分区）处理，这样作的目的就是为了分布式。

你可让这个RDD有两个分区，那么有多是这个形式：RDD(1,2) (3,4)。

这样设计的目的在于：能够进行分布式运算。

注：建立RDD的方式有多种，好比案例一中是基于一个基本的集合类型（Array）转换而来，像parallelize这样的方法还有不少，以后就会学到。此外，咱们也能够在读取数据集时就建立RDD。

案例二

Spark can create distributed datasets from any storage source supported by Hadoop, including your local file system, HDFS, Cassandra, HBase, Amazon S3, etc. Spark supports text files, SequenceFiles, and any other Hadoop InputFormat.

Text file RDDs can be created using SparkContext’s textFile method. This method takes an URI for the file (either a local path on the machine, or a hdfs://, s3n://, etc URI) and reads it as a collection of lines. Here is an example invocation:

val distFile = sc.textFile("data.txt")

查看RDD

rdd.collect

收集rdd中的数据组成Array返回，此方法将会把分布式存储的rdd中的数据集中到一台机器中组建Array。

在生产环境下必定要慎用这个方法，容易内存溢出。

查看RDD的分区数量：

rdd.partitions.size

查看RDD每一个分区的元素：

rdd.glom.collect

此方法会将每一个分区的元素以Array形式返回。

2．分区概念

在上图中，一个RDD有item1~item25个数据，共5个分区，分别在3台机器上进行处理。此外，spark并无原生的提供rdd的分区查看工具，咱们能够本身来写一个。

示例代码：

import org.apache.spark.rdd.RDD
import scala.reflect.ClassTag
object su {
 def debug[T: ClassTag](rdd: RDD[T]) = {
  rdd.mapPartitionsWithIndex((i: Int, iter: Iterator[T]) => {
   val m = scala.collection.mutable.Map[Int, List[T]]()
   var list = List[T]()
   while (iter.hasNext) {
    list = list :+ iter.next
   }
   m(i) = list
   m.iterator
  }).collect().foreach((x: Tuple2[Int, List[T]]) => {
   val i = x._1
   println(s"partition:[$i]")
   x._2.foreach { println }
  })
 }
}

3．RDD操做

针对RDD的操做，分两种，一种是Transformation（变换），一种是Actions（执行）。

Transformation（变换）操做属于懒操做，不会真正触发RDD的处理计算。

Actions（执行）操做才会真正触发。

1>Transformations

①map(func)

Return a new distributed dataset formed by passing each element of the source through a function func.

参数是函数，函数应用于RDD每个元素，返回值是新的RDD。

案例展现：

map将函数应用到rdd的每一个元素中。

val rdd = sc.makeRDD(List(1,3,5,7,9))
rdd.map(_*10)

②flatMap(func)

Similar to map, but each input item can be mapped to 0 or more output items (so func should return a Seq rather than a single item).

扁平化map，对RDD每一个元素转换，而后再扁平化处理。

案例展现：

flatMap 扁平map处理

val rdd = sc.makeRDD(List("hello world","hello count","world spark"),2)
rdd.map(_.split{" "})//Array(Array(hello, world), Array(hello, count), Array(world, spark))
rdd.flatMap(_.split{" "})//Array[String] = Array(hello, world, hello, count, world, spark)
//Array[String] = Array(hello, world, hello, count, world, spark)

注：map和flatMap有何不一样？

map：对RDD每一个元素转换。

flatMap：对RDD每一个元素转换，而后再扁平化（即去除集合）

因此，通常咱们在读取数据源后，第一步执行的操做是flatMap。

③filter(func)

Return a new dataset formed by selecting those elements of the source on which func returns true.

参数是函数，函数会过滤掉不符合条件的元素，返回值是新的RDD。

案例展现：

filter用来从rdd中过滤掉不符合条件的数据。

val rdd = sc.makeRDD(List(1,3,5,7,9))
rdd.filter(_<5)

④mapPartitions(func)

Similar to map, but runs separately on each partition (block) of the RDD, so func must be of type Iterator<T> => Iterator<U> when running on an RDD of type T.

该函数和map函数相似，只不过映射函数的参数由RDD中的每个元素变成了RDD中每个分区的迭代器。

案例展现：

scala>val rdd3 = rdd1.mapPartitions{ x => {
      val result = List[Int]()
      var i = 0
      while(x.hasNext){
      i += x.next()
      }
      result.::(i).iterator
      }}
scala>rdd3.collect

⑤mapPartitionsWithIndex(func)

Similar to mapPartitions, but also provides func with an integer value representing the index of the partition, so func must be of type (Int, Iterator<T>) => Iterator<U> when running on an RDD of type T.

函数做用同mapPartitions，不过提供了两个参数，第一个参数为分区的索引。

案例展现：

var rdd1 = sc.makeRDD(1 to 5,2)
var rdd2 = rdd1.mapPartitionsWithIndex{
(index,iter) => {
var result = List[String]()
var i = 0
while(iter.hasNext){
i += iter.next()
}
result.::(index + "|" + i).iterator
}
}

案例展现：

val rdd = sc.makeRDD(List(1,2,3,4,5),2)
rdd.mapPartitionsWithIndex((index,iter)=>{
var list = List[String]()
while(iter.hasNext){
if(index==0)
list = list :+ (iter.next + "a")
else {
list = list :+ (iter.next + "b")
}
}
list.iterator
})

⑥union(otherDataset)

Return a new dataset that contains the union of the elements in the source dataset and the argument.

union并集，也能够用++实现。

案例展现：

val rdd1 = sc.makeRDD(List(1,3,5))
val rdd2 = sc.makeRDD(List(2,4,6,8))
val rdd = rdd1.union(rdd2)
val rdd = rdd1 ++ rdd2

⑦intersection(otherDataset)

Return a new RDD that contains the intersection of elements in the source dataset and the argument.

intersection交集

案例展现：

val rdd1 = sc.makeRDD(List(1,3,5,7))
val rdd2 = sc.makeRDD(List(5,7,9,11))
val rdd = rdd1.intersection(rdd2)

⑧subtract

求差集。

案例展现：

val rdd1 = sc.makeRDD(List(1,3,5,7,9))
val rdd2 = sc.makeRDD(List(5,7,9,11,13))
val rdd =  rdd1.subtract(rdd2)

⑨distinct([numTasks])

Return a new dataset that contains the distinct elements of the source dataset.

没有参数，将RDD里的元素进行去重操做。

案例展现：

val rdd = sc.makeRDD(List(1,3,5,7,9,3,7,10,23,7))
rdd.distinct

⑩groupByKey([numTasks])

When called on a dataset of (K, V) pairs, returns a dataset of (K, Iterable<V>) pairs.

Note:If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using reduceByKey or aggregateByKey will yield much better performance.

Note:By default, the level of parallelism in the output depends on the number of partitions of the parent RDD. You can pass an optional numTasks argument to set a different number of tasks.

groupByKey对于数据格式是有要求的，即操做的元素必须是一个二元tuple，tuple._1是key，tuple._2是value。

好比下面两种种数据格式都不符合要求：

sc.parallelize(List("dog", "tiger", "lion", "cat", "spider", "eagle"), 2)
sc.parallelize(List(("cat",2,1),("dog",5,1),("cat",4,1),("dog",3,2),
                    ("cat",6,2),("dog",3,4),("cat",9,4),("dog",1,4)),2)

案例展现：

scala>val rdd = sc.parallelize(List(("cat",2), ("dog",5),("cat",4),
                                    ("dog",3),("cat",6),("dog",3),
                                    ("cat",9),("dog",1)),2)
scala>rdd.groupByKey()

⑪reduceByKey(func, [numTasks])

When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function func, which must be of type (V,V) => V. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument.

reduceByKey操做的数据格式必须是一个二元tuple

案例展现：

scala>var rdd = sc.makeRDD( List( ("hello",1),("spark",1),("hello",1),("world",1) ) )
rdd.reduceByKey(_+_);

⑫aggregateByKey(zeroValue)(seqOp,combOp,[numTasks])

When called on a dataset of (K, V) pairs, returns a dataset of (K, U) pairs where the values for each key are aggregated using the given combine functions and a neutral "zero" value. Allows an aggregated value type that is different than the input value type, while avoiding unnecessary allocations. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument.

aggregateByKey(zeroValue)(func1,func2)

zeroValue表示初始值，初始值会参与func1的计算，在分区内，按key分组，把每组的值进行fun1的计算，再将每一个分区每组的计算结果按fun2进行计算

scala> val rdd = sc.parallelize(List(("cat",2),("dog",5),("cat",4),("dog",3),("cat",6),("dog",3),("cat",9),("dog",1)),2);

查看分区结果：

partition:[0]
(cat,2)
(dog,5)
(cat,4)
(dog,3)
partition:[1]
(cat,6)
(dog,3)
(cat,9)
(dog,1)
scala> rdd.aggregateByKey(0)(_+_,_+_);

scala> rdd.aggregateByKey(0)(_+_,_*_);

⑬sortByKey([ascending],[numTasks])

When called on a dataset of (K, V) pairs where K implements Ordered, returns a dataset of (K, V) pairs sorted by keys in ascending or descending order, as specified in the boolean ascending argument.

案例展现：

val d2 = sc.parallelize(Array(("cc",32),("bb",32),("cc",22),("aa",18),("bb",6),("dd",16),("ee",104),("cc",1),("ff",13),("gg",68),("bb",44)))
d2.sortByKey(true).collect

⑭join(otherDataset,[numTasks])

When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (V, W)) pairs with all pairs of elements for each key. Outer joins are supported through leftOuterJoin,rightOuterJoin, and fullOuterJoin.

案例展现：

val rdd1 = sc.makeRDD(List(("cat",1),("dog",2)))
val rdd2 = sc.makeRDD(List(("cat",3),("dog",4),("tiger",9)))
rdd1.join(rdd2);

⑮cartesian(otherDataset)

When called on datasets of types T and U, returns a dataset of (T, U) pairs (all pairs of elements).

参数是RDD，求两个RDD的笛卡儿积。

案例展现：

cartesian 笛卡尔积

val rdd1 = sc.makeRDD(List(1,2,3))
val rdd2 = sc.makeRDD(List("a","b"))
rdd1.cartesian(rdd2);

⑯coalesce(numPartitions)

Decrease the number of partitions in the RDD to numPartitions. Useful for running operations more efficiently after filtering down a large dataset.

coalesce(n,true/false)扩大或缩小分区。

案例展现：

val rdd = sc.makeRDD(List(1,2,3,4,5),2)
rdd9.coalesce(3,true);

若是是扩大分区，须要传入一个true，表示要从新shuffle。

rdd9.coalesce(2);

若是是缩小分区，默认就是false，不须要明确的传入。

⑰repartition(numPartitions)

Reshuffle the data in the RDD randomly to create either more or fewer partitions and balance it across them. This always shuffles all data over the network.

repartition(n) 等价于上面的coalesce

⑱partitionBy

一般在建立RDD时指定分区规则，将会致使数据自动分区；也能够经过partitionBy方法人为指定分区方式来进行分区。

常见的分区器有：

HashPartitioner、RangePartitioner

案例展现：

import org.apache.spark._
val r1 = sc.makeRDD(List((2,"aaa"),(9,"bbb"),(7,"ccc"),(9,"ddd"),(3,"eee"),(2,"fff")),2);
val r2=r1.partitionBy(new HashPartitioner(2))

按照键的hash%分区数获得的编号去往指定的分区，这种方式能够实现将相同键的数据分发给同一个分区的效果。

val r3=r1.partitionBy(new RangePartitioner(2,r1))

将数据按照值的字典顺序进行排序，再分区。

2>Actions

①reduce(func)

Aggregate the elements of the dataset using a function func (which takes two arguments and returns one). The function should be commutative and associative so that it can be computed correctly in parallel.

并行整合全部RDD数据，例如求和操做。

②collect

Return all the elements of the dataset as an array at the driver program. This is usually useful after a filter or other operation that returns a sufficiently small subset of the data.

返回RDD全部元素，将rdd分布式存储在集群中不一样分区的数据获取到一块儿组成一个数组返回。

要注意：这个方法将会把全部数据收集到一个机器内，容易形成内存的溢出，在生产环境下千万慎用。

③count

Return the number of elements in the dataset.

统计RDD里元素个数

案例展现：

val rdd = sc.makeRDD(List(1,2,3,4,5),2)
rdd.count

④first

Return the first element of the dataset (similar to take(1)).

⑤take(n)

Return an array with the first n elements of the dataset.

take获取前n个数据。

案例展现：

val rdd = sc.makeRDD(List(52,31,22,43,14,35))
rdd.take(2)

⑥takeOrdered(n,[ordering])

Return the first n elements of the RDD using either their natural order or a custom comparator.

takeOrdered(n)先将对象中的数据进行升序排序，而后取前n个。

案例展现：

val rdd = sc.makeRDD(List(52,31,22,43,14,35))
rdd.takeOrdered(3)

⑦top(n)

top(n)先将对象中的数据进行降序排序，而后取前n个。

val rdd = sc.makeRDD(List(52,31,22,43,14,35))
rdd.top(3)

⑧saveAsTextFile(path)

Write the elements of the dataset as a text file (or set of text files) in a given directory in the local filesystem, HDFS or any other Hadoop-supported file system. Spark will call toString on each element to convert it to a line of text in the file.

saveAsTextFile 按照文本方式保存分区数据，到指定路径。

案例示例：

val rdd = sc.makeRDD(List(1,2,3,4,5),2);
rdd.saveAsTextFile("/root/work/aaa")

⑨countByKey()

Only available on RDDs of type (K, V). Returns a hashmap of (K, Int) pairs with the count of each key.

⑩foreach(func)

Run a function func on each element of the dataset. This is usually done for side effects such as updating an Accumulator or interacting with external storage systems.

Note: modifying variables other than Accumulators outside of the foreach() may result in undefined behavior. See Understanding closures for more details.

3>案例

经过rdd实现统计文件中的单词数量。

sc.textFile("/root/work/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("/root/work/wcresult")

4．RDD的依赖关系

RDD之间的关系能够从两个维度来理解：

一个是RDD是从哪些RDD转换而来，也就是RDD的parent RDD（s）是什么；还有就是依赖于parent RDD（s）的哪些Partition（s）。这个关系，就是RDD之间的依赖，org.apache.spark.Dependency。

根据依赖于parent RDD（s）的Partitions的不一样状况，Spark将这种依赖分为两种，一种是宽依赖，一种是窄依赖。

RDD和它依赖的parent RDD（s）的关系有两种不一样的类型，即窄依赖（narrow dependency）和宽依赖（wide dependency）。

窄依赖指的是每个parent RDD的Partition最多被子RDD的一个Partition使用，以下图所示。

宽依赖指的是多个子RDD的Partition会依赖同一个parent RDD的Partition。

咱们能够从不一样类型的转换来进一步理解RDD的窄依赖和宽依赖的区别，以下图所示。

1>窄依赖

对于窄依赖操做，它们只是将Partition的数据根据转换的规则进行转化，并不涉及其余的处理，能够简单地认为只是将数据从一个形式转换到另外一个形式。

窄依赖底层的源码：

abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
 //返回子RDD的partitionId依赖的全部的parent RDD的Partition(s)
 def getParents(partitionId: Int): Seq[Int]
  override def rdd: RDD[T] = _rdd
 }
 class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
  override def getParents(partitionId: Int) = List(partitionId)
}

因此对于窄依赖，并不会引入昂贵的Shuffle。因此执行效率很是高。若是整个DAG中存在多个连续的窄依赖，则能够将这些连续的窄依赖整合到一块儿连续执行，中间不执行shuffle 从而提升效率，这样的优化方式称之为流水线优化。

此外，针对窄依赖，若是子RDD某个分区数据丢失，只须要找到父RDD对应依赖的分区，恢复便可。但若是是宽依赖，当分区丢失时，最糟糕的状况是要重算全部父RDD的全部分区。

2>宽依赖

对于groupByKey这样的操做，子RDD的全部Partition（s）会依赖于parent RDD的全部Partition（s），子RDD的Partition是parent RDD的全部Partition Shuffle的结果。

宽依赖的源码：

class ShuffleDependency[K, V, C](
    @transient _rdd: RDD[_ <: Product2[K, V]],
    val partitioner: Partitioner,
    val serializer: Option[Serializer] = None,
    val keyOrdering: Option[Ordering[K]] = None,
    val aggregator: Option[Aggregator[K, V, C]] = None,
    val mapSideCombine: Boolean = false)
extends Dependency[Product2[K, V]] {
 
override def rdd = _rdd.asInstanceOf[RDD[Product2[K, V]]]
//获取新的shuffleId
val shuffleId: Int = _rdd.context.newShuffleId()
//向ShuffleManager注册Shuffle的信息
val shuffleHandle: ShuffleHandle =
_rdd.context.env.shuffleManager.registerShuffle(
    shuffleId, _rdd.partitions.size, this)
 
    _rdd.sparkContext.cleaner.foreach(_.registerShuffleForCleanup(this))
}

spark中一旦遇到宽依赖就须要进行shuffle的操做，所谓的shuffle的操做的本质就是将数据汇总后从新分发的过程。

这个过程数据要汇总到一块儿，数据量可能很大因此不可避免的须要进行数据落磁盘的操做，会下降程序的性能，因此spark并非彻底内存不读写磁盘，只能说它尽力避免这样的过程来提升效率。

5．RDD容错机制

分布式系统一般在一个机器集群上运行，同时运行的几百台机器中某些出问题的几率大大增长，因此容错设计是分布式系统的一个重要能力。

Spark之前的集群容错处理模型，像MapReduce，将计算转换为一个有向无环图（DAG）的任务集合，这样能够经过重复执行DAG里的一部分任务来完成容错恢复。可是因为主要的数据存储在分布式文件系统中，没有提供其余存储的概念，容错过程须要在网络上进行数据复制，从而增长了大量的消耗。因此，分布式编程中常常须要作检查点，即将某个时机的中间数据写到存储（一般是分布式文件系统）中。

RDD也是一个DAG，每个RDD都会记住建立该数据集须要哪些操做，跟踪记录RDD的继承关系，这个关系在Spark里面叫lineage（血缘关系）。当一个RDD的某个分区丢失时，RDD是有足够的信息记录其如何经过其余RDD进行计算，且只需从新计算该分区，这是Spark的一个创新。

6．RDD的缓存

相比Hadoop MapReduce来讲，Spark计算具备巨大的性能优点，其中很大一部分缘由是Spark对于内存的充分利用，以及提供的缓存机制。

持久化在早期被称做缓存（cache），但缓存通常指将内容放在内存中。虽然持久化操做在绝大部分状况下都是将RDD缓存在内存中，但通常都会在内存不够时用磁盘顶上去（比操做系统默认的磁盘交换性能高不少）。固然，也能够选择不使用内存，而是仅仅保存到磁盘中。因此，如今Spark使用持久化（persistence）这一更普遍的名称。

若是一个RDD不止一次被用到，那么就能够持久化它，这样能够大幅提高程序的性能，甚至达10倍以上。

默认状况下，RDD只使用一次，用完即扔，再次使用时须要从新计算获得，而持久化操做避免了这里的重复计算，实际测试也显示持久化对性能提高明显，这也是Spark刚出现时被人称为内存计算框架的缘由。

假设首先进行了RDD0→RDD1→RDD2的计算做业，那么计算结束时，RDD1就已经缓存在系统中了。在进行RDD0→RDD1→RDD3的计算做业时，因为RDD1已经缓存在系统中，所以RDD0→RDD1的转换不会重复进行，计算做业只须进行RDD1→RDD3的计算就能够了，所以计算速度能够获得很大提高。

1>持久化等级

持久化的方法是调用persist()函数，除了持久化至内存中，还能够在persist()中指定storage level参数使用其余的类型，具体以下：

①MEMORY_ONLY

MEMORY_ONLY：将RDD以反序列化的Java对象的形式存储在JVM中。若是内存空间不足，部分数据分区将不会被缓存，在每次须要用到这些数据时从新进行计算。这是默认的级别。

persist()方法的默认级别就是cache()方法，cache()方法对应的级别就是MEMORY_ONLY级别。

②MEMORY_AND_DISK

MEMORY_AND_DISK：将RDD以反序列化的Java对象的形式存储在JVM中。若是内存空间不够，将未缓存的数据分区存储到磁盘，在须要使用这些分区时从磁盘读取，存入磁盘的对象也是没有通过序列化的。

③MEMORY_ONLY_SER

MEMORY_ONLY_SER：将RDD以序列化的Java对象的形式进行存储（每一个分区为一个byte数组）。这种方式会比反序列化对象的方式节省不少空间，尤为是在使用fast serialize时会节省更多的空间，可是在读取时会使得CPU的read变得更加密集。若是内存空间不够，部分数据分区将不会被缓存，在每次须要用到这些数据时从新进行计算。

④MEMORY_AND_DISK_SER

MEMORY_AND_DISK_SER：相似于MEMORY_ONLY_SER，可是溢出的分区会存储到磁盘，而不是在用到它们时从新计算，存储到磁盘上的对象会进行序列化。在须要使用这些分区时从磁盘读取。

⑤DISK_ONLY

DISK_ONLY：只在磁盘上缓存RDD。

⑥MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.

MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc. ：与上面的级别功能相同，只不过每一个分区在集群中两个节点上创建副本。

⑦OFF_HEAP

OFF_HEAP：将数据存储在off-heap memory中。使用堆外内存，这是Java虚拟机里面的概念，堆外内存意味着把内存对象分配在Java虚拟机的堆之外的内存，这些内存直接受操做系统管理（而不是虚拟机）。注意，可能带来一些GC回收问题。

Spark也会自动持久化一些在shuffle操做过程当中产生的临时数据（好比reduceByKey），即使是用户并无调用持久化的方法。这样作能够避免当shuffle阶段时若是一个节点挂掉了就得从新计算整个数据的问题。若是用户打算屡次重复使用这些数据，咱们仍然建议用户本身调用持久化方法对数据进行持久化。

2>使用缓存

须要先导包，而后才能调用命令。

scala> import org.apache.spark.storage._
scala> val rdd1=sc.makeRDD(1 to 5)
scala> rdd1.cache  //cache只有一种默认的缓存级别，即MEMORY_ONLY
scala> rdd1.persist(StorageLevel.MEMORY_ONLY)

3>缓存数据的清除

Spark会自动监控每一个节点上的缓存数据，而后使用least-recently-used(LRU)机制来处理旧的缓存数据。若是你想手动清理这些缓存的RDD数据而不是去等待它们被自动清理掉，

可使用RDD.unpersist()方法。

cala> rdd1.unpersist()

二、DAG

Spark会根据用户提交的计算逻辑中的RDD的转换和动做来生成RDD之间的依赖关系，同时这个计算链也就生成了逻辑上的DAG。接下来以“Word Count”为例，详细描述这个DAG生成的实现过程。

Spark Scala版本的Word Count程序以下：

1：val file=sc.textFile("hdfs://hadoop01:9000/hello1.txt")
2：val counts = file.flatMap(line => line.split(" "))
3： .map(word=>(word,1))
4： .reduceByKey(_+_)
5：counts.saveAsTextFile("hdfs://...")

file和counts都是RDD，其中file是从HDFS上读取文件并建立了RDD，而counts是在file的基础上经过flatMap、map和reduceByKey这三个RDD转换生成的。最后，counts调用了动做saveAsTextFile，用户的计算逻辑就从这里开始提交的集群进行计算。那么上面这5行代码的具体实现是什么呢？

行1：sc是org.apache.spark.SparkContext的实例，它是用户程序和Spark的交互接口，会负责链接到集群管理者，并根据用户设置或者系统默认设置来申请计算资源，完成RDD的建立等。

sc.textFile（"hdfs：//..."）就完成了一个org.apache.spark.rdd.HadoopRDD的建立，而且完成了一次RDD的转换：经过map转换到一个org.apache.spark.rdd.MapPartitions-RDD。也就是说，file其实是一个MapPartitionsRDD，它保存了文件的全部行的数据内容。

行2：将file中的全部行的内容，以空格分隔为单词的列表，而后将这个按照行构成的单词列表合并为一个列表。最后，以每一个单词为元素的列表被保存到MapPartitionsRDD。

行3：将第2步生成的MapPartittionsRDD再次通过map将每一个单词word转为（word，1）的元组。这些元组最终被放到一个MapPartitionsRDD中。

行4：首先会生成一个MapPartitionsRDD，起到map端combiner的做用；而后会生成一个ShuffledRDD，它从上一个RDD的输出读取数据，做为reducer的开始；最后，还会生成一个MapPartitionsRDD，起到reducer端reduce的做用。

行5：向HDFS输出RDD的数据内容。最后，调用org.apache.spark.SparkContext#runJob向集群提交这个计算任务。

1．DAG的产生

原始的RDD（s）经过一系列转换就造成了DAG。RDD之间的依赖关系，包含了RDD由哪些Parent RDD（s）转换而来和它依赖parent RDD（s）的哪些Partitions，是DAG的重要属性。

借助这些依赖关系，DAG能够认为这些RDD之间造成了Lineage（血统，血缘关系）。借助Lineage，能保证一个RDD被计算前，它所依赖的parent RDD都已经完成了计算；同时也实现了RDD的容错性，即若是一个RDD的部分或者所有的计算结果丢失了，那么就须要从新计算这部分丢失的数据。

2．Stage

Spark在执行任务（job）时，首先会根据依赖关系，将DAG划分为不一样的阶段（Stage）。

处理流程是：

1）Spark在执行Transformation类型操做时都不会当即执行，而是懒执行（计算）。

2）执行若干步的Transformation类型的操做后，一旦遇到Action类型操做时，才会真正触发执行（计算）。

3）执行时，从当前Action方法向前回溯，若是遇到的是窄依赖则应用流水线优化，继续向前找，直到碰到某一个宽依赖。

4）由于宽依赖必需要进行shuffle，没法实现优化，因此将这一次段执行过程组装为一个stage。

5）再从当前宽依赖开始继续向前找。重复刚才的步骤，从而将这个DAG还分为若干的stage。

在stage内部能够执行流水线优化，而在stage之间没办法执行流水线优化，由于有shuffle。可是这种机制已经尽力的去避免了shuffle。

3．Job和Task

原始的RDD通过一系列转换后（一个DAG），会在最后一个RDD上触发一个动做，这个动做会生成一个Job。

因此能够这样理解：一个DAG对应一个Spark的Job。

在Job被划分为一批计算任务（Task）后，这批Task会被提交到集群上的计算节点去计算。

Spark的Task分为两种：

1）org.apache.spark.scheduler.ShuffleMapTask

2）org.apache.spark.scheduler.ResultTask

简单来讲，DAG的最后一个阶段会为每一个结果的Partition生成一个ResultTask，其他全部的阶段都会生成ShuffleMapTask。

三、可视化理解窄依赖和宽依赖

案例单词统计

scala>val data=sc.textFile("/home/software/hello.txt",2)
scala> data.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

1）打开web页面控制台（ip:4040端口地址），刷新，会发现刚才的操做会出如今页面上。

2）点击 Description下的 collect at…… 进入job的详细页面

3）点击 DAG Visualization 会出现以下图形

四、综合案例

1．WordCount

数据样例：

hello scala
hello spark
hello world

1>导入jar包

建立spark的项目，在scala中建立项目，导入spark相关的jar包。

2>代码示例

统计单词个数，并输出

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object Driver {
  
  def main(args: Array[String]): Unit = {
    //建立Spark的环境对象，配置运行模式
    //1：集群模式:setMaster("spark://yun01:7077")
    //2：本地模式:setMaster("local")
    val conf=new SparkConf().setMaster("spark://yun01:7077").setAppName("wordcount")
    //获取Spark上下文对象
    val sc=new SparkContext(conf)
    val data=sc.textFile("hdfs://yun01:9000/words.txt", 2)
    val result=data.flatMap { x => x.split(" ") }.map { x => (x,1) }.reduceByKey(_+_)
    result.saveAsTextFile("hdfs://yun01:9000/wcresult")
  }
}

3>服务器运行

将写好的项目打成jar，上传到服务器，进入bin目录，执行以下命令：

spark-submit --class cn.tedu.WordCountDriver /home/software/spark/conf/wc.jar

注意输出的目录必须是不存在的，若是存在会报错。

2．求平均值

数据样例：

第一列是编号，第二列是数据。

正确答案：42

1>分析

①只拿第二列，造成RDD

②类型转换->String->Int

③和/个数

2>代码示例一

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object Average {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setMaster("local").setAppName("average")
    val sc=new SparkContext(conf)
    val data=sc.textFile("d://data/average.txt")
    val r11=data.map { line => line.split(" ")(1).toInt }.reduce(_+_)
    val r1=data.flatMap { x => x.split(" ").drop(1) }.map { x => x.toInt }.reduce(_+_)
    val r2=data.count()
    val result=r1/r2
    println(result)
  }
}

3>代码示例二

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object AverageDriver {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("AverageDriver")
    val sc = new SparkContext(conf)
    val data = sc.textFile("d://average.txt", 3)
    val ageData = data.map { line => { line.split(" ")(1).toInt } }
    val ageSum = ageData.mapPartitions { it =>
      {
        val result = List[Int]()
        var i = 0
        while (it.hasNext) {
          i += it.next()
        }
        result.::(i).iterator
      }
    }.reduce(_ + _)
    val pepopleCount = data.count()
    val average = ageSum / pepopleCount
    println(average)
  }
}

3．求最大值和最小值

数据样例：

第一列是编号，第二列是性别，第三列是身高。

1>代码示例

获取最大、最小值。

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import breeze.linalg.split
import scala.collection.Iterable
object MaxMin {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("Max")
    val sc = new SparkContext(conf)
    val data = sc.textFile("d://data/MaxMin.txt")
    //第一种方法
    val r1 = data.filter { line => line.contains("M") }
                 .map { line => line.split(" ")(2).toInt }
    //第二种方法
    val r2 = data.filter { line => line.split(" ")(1)
                 .equals("M") }.map { line => line.split(" ")(2).toInt }
    val max = r1.max()
    val min = r1.min()
    println(max + " " + min)
  }
}

2>代码示例

获取最大、最小值的所有信息。

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object MaxMinInfo {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setMaster("local").setAppName("Info")
    val sc=new SparkContext(conf)
    val data=sc.textFile("d://data/MaxMin.txt")
    
    val r1=data.filter { line => line.split(" ")(1)
               .equals("M") }
               .map { x => (x.split(" ")(0),x.split(" ")(1),x.split(" ")(2).toInt) }
    
    val max=r1.sortBy(x=> -x._3, true, 1).take(1)
    val min=r1.sortBy(x=> x._3).take(1)
    
    println(max(0)._1+" "+max(0)._2+" "+max(0)._3)
    println(min(0)._1+" "+min(0)._2+" "+min(0)._3)
  }
}

4．TopK

统计单词出现的次数最多的前三个。

1>数据样例

hello world bye world
hello hadoop bye hadoop
hello world java web
hadoop scala java hive
hadoop hive redis hbase
hello hbase java redis

2>分析

Top K算法有两步，一是统计词频，二是找出词频最高的前K个词。

3>代码示例

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object Topk {
  def main(args: Array[String]): Unit = {
    
    val conf=new SparkConf().setMaster("local").setAppName("topk")
    val sc=new SparkContext(conf)
    val data=sc.textFile("d://data/topk.txt")
    //方法一
    val r1=data.flatMap { _.split(" ").map { x => (x,1) } }
               .reduceByKey(_+_).sortBy(x=> -x._2).take(3)
    r1.foreach(println(_))
    //方法二
    val r2=data.flatMap { x => x.split(" ") }.groupBy { x => x }
               .map{x=>(x._1,x._2.count { x => true })}
               .map{case(word,count)=>(count,word)}.top(3)
    r2.foreach(println(_))
    //方法三
    val wordcunt=data.flatMap { x => x.split(" ")
                     .map { x => (x,1) } }.reduceByKey(_+_)
    val top3=wordcunt.top(3)(Ordering.by { case(word,count) => count })
    top3.foreach(println(_))
  }
}

4>应用场景

Top K的示例模型能够应用在求过去一段时间消费次数最多的消费者、访问最频繁的IP地址和最近、更新、最频繁的微博等应用场景。

5．求中位数

数据样例：

1 20 8 2 5 11 29 10
7 4 45 6 23 17 19

一共是15个数，正确答案是10

代码示例

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.dmg.pmml.True

object Median {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setMaster("local").setAppName("median")
    val sc=new SparkContext(conf)
    val data=sc.textFile("d://data/median.txt")
    //方法一
    val r1=data.flatMap { x => x.split(" ") }.count().toInt
    val zhi=r1/2
    val r2=data.flatMap { x => x.split(" ") }
               .map { x => x.toInt }
               .sortBy(x=>x, true, 1).take(r1)(zhi)
    println(r2)
    //方法二
    val sortData=data.flatMap { x => x.split(" ") }
                     .map { x => x.toInt }.sortBy(x=>x)
                     .take(zhi+1).last
    println(sortData)
  }
}

6．二次排序

数据样例：

aa 12
bb 32
aa 3
cc 43
dd 23
cc 5
cc 8
bb 33
bb 12

要求：先按第一例升序排序，再按第二列降序排序。

1>代码示例

①自定义排序类

import scala.math.Ordered

class SecondarySort(v1:String,v2:Int) extends Ordered[SecondarySort] with Serializable {
  var col1=v1
  var col2=v2
  def compare(that: SecondarySort): Int = {
    //按第一列作升序排序
    val tmp=this.col1.compareTo(that.col1)
    if(tmp==0){
      //按第二列作降序排序
      that.col2.compareTo(this.col2)
    }else{
      tmp
    }
  }
}

②Driver

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object Driver {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setMaster("local").setAppName("ssort")
    val sc=new SparkContext(conf)
    val data=sc.textFile("d://data/ssort.txt")
    //用sortByKey来实现二次排序，因此先把数据组成一个二元Tuple
    //二元Tuple的形式（SecondarySort（col1，col2），line）
    val result=data.map { line => 
      val infos=line.split(" ")
      (new SecondarySort(infos(0),infos(1).toInt),line)
    }.sortByKey().map(x=>x._2)
    result.foreach(println(_))
  }
}

7．倒排索引

数据样例：

doc1.txt:
hello spark
hello hadoop

doc2.txt:
hello hive
hello hbase
hello spark

doc3.txt:
hadoop hbase
hive scala

最后的结果形式为：

1>代码示例

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object Driver {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setMaster("local").setAppName("invert")
    val sc=new SparkContext(conf)
    //将指定目录下的全部文件，返回到一个RDD中
    val data=sc.wholeTextFiles("d://data/inverted/*")
    data.foreach(println(_))
    //且分时，先按\r\n，而后按空格切
    val result=data.map{case(filePath,text)=>
      //切分出文件名称
      val fileName=filePath.split("/").last.dropRight(4)
      (fileName,text)
    }.flatMap{case(filename,text)=>
      //切分文件内容
      text.split("\r\n").flatMap { line => line.split(" ") }
      //调换内容和文件名称的位置
      .map { word => (word,filename) } 
    }
    //经过单词分组
    .groupByKey()
    //聚合文件名称
    .map{case(word,buffer)=>(word,buffer.toSet.mkString(","))}
    result.foreach(println)
    
  }
}

上一篇：Spark On Yarn彻底分布式搭建

下一篇：Spark的架构