spark相关概念

RDD

Resilient Distributed Dataset (RDD)弹性分布数据集,是对Spark分布式内存的抽象,RDD实现了以操作本地集合的方式来操作分布式数据集的抽象实现。任何数据在Spark中都被表示为RDD,RDD可以cache到内存中,每次对RDD数据集的操作之后的结果,都可以存放到内存中,下一个操作可以直接从内存中输入,省去了MapReduce大量的磁盘IO操作。这对于迭代运算比较常见的机器学习算法, 交互式数据挖掘来说,效率提升比较大。从编程的角度来看,RDD可以简单看成是一个数组。和普通数组的区别是,RDD中的数据是分区存储的,这样不同分区的数据就可以分布在不同的机器上,同时可以被并行处理。因此,Spark应用程序所做的无非是把需要处理的数据转换为RDD,然后对RDD进行一系列的变换和操作从而得到结果。

可以将RDD理解为一个具有容错机制的特殊集合,它提供了一种只读、只能有已存在的RDD变换而来的共享内存,然后将所有数据都加载到内存中,方便进行多次重用。a.他是分布式的,可以分布在多台机器上,进行计算。b.他是弹性的,计算过程中内错不够时它会和磁盘进行数据交换。c.这些限制可以极大的降低自动容错开销d.实质是一种更为通用的迭代并行计算框架,用户可以显示的控制计算的中间结果,然后将其自由运用于之后的计算。

RDD容错机制

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中的所有数据块全部重新计算来恢复。所以在长“血统”链特别是有宽依赖的时候,需要在适当的时机设置数据检查点。也是这两个特性要求对于不同依赖关系要采取不同的任务调度机制和容错恢复机制。

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则是这个抽象方法的实现。

如何操作RDD?

(1)如何获取RDD
a.从共享的文件系统获取(如:HDFS)
b.通过已存在的RDD转换
c.将已存在scala集合(只要是Seq对象)并行化 ,通过调用SparkContext的parallelize方法实现
d.改变现有RDD的之久性;RDD是懒散,短暂的。(RDD的固化:cache缓存至内存; save保存到分布式文件系统)

(2)操作RDD的两个操作(operation)
a.Actions:对数据集计算后返回一个数值value给驱动程序;例如:Reduce将数据集的所有元素用某个函数聚合后,将最终结果返回给程序。
b.Transformation:根据数据集创建一个新的数据集,计算后返回一个新RDD,Transformation不会改变原始RDD;例如:Map将数据的每个元素经过某个函数计算后,返回一个姓的分布式数据集。
区分Action和Transformation的简单方法是:查看返回值,如果返回值RDD,那么就是Transformation,如果是其他返回值的话,就是Action。
Transformation的执行都是lazy的,在Action执行之前,对RDD做的Transformation操作都只会记录操作指令,而不会真的去执行操作,直到操作序列中出现Action才会触发Transformation的执行。

Lazy Evaluation

Transformation的执行都是lazy的,在Action执行之前,对RDD做的Transformation操作都只会记录操作序列,而不会真的去执行操作,直到操作序列中出现Action才会触发Transformation的执行。
Lazy Evaluation能够有效减少加载RDD的次数,Hadoop MapReduce就需要程序员来把操作聚合在一起以便能够尽量减少数据的加载次数。

Pass function to operation

对于传入Action和Transformation的函数,比如:

class SearchFunctions(val query:string){
    def isMatch(s:String):Boolean = {
        s.contains(s)
    }
    def getMatchesFunctionsRefrence(rdd:RDD[string]): RDD[string]={
        rdd.map(isMatch)
    }
    def getMatchesNoRefrence(rdd:RDD[string]): RDD[string]={
        val isMatch = this.isMatch
        rdd.map(isMatch)
    }
}

如果将getMatchesFunctionsRefrence传递给operation的话,那么整个class SearchFunctions传递给计算节点,因为getMatchesFunctionsRefrence中引用了class的私有函数isMatch,这不仅造成传输量大的问题,还有可能造成work node上的代码执行失败;正确的做法是将私有成员转换成函数内部常量,比如getMatchesNoReference

Actions

reduce(func) 通过函数func聚集数据集中的所有元素。Func函数接受2个参数,返回一个值。这个函数必须是关联性的,确保可以被正确的并发执行

collect() 在Driver的程序中,以数组的形式,返回数据集的所有元素。这通常会在使用filter或者其它操作后,返回一个足够小的数据子集再使用,直接将整个RDD集Collect返回,很可能会让Driver程序OOM,因为collect()会将RDD加载到内存。

count() 返回数据集的元素个数

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。这通常用于更新一个累加器变量,或者和外部存储系统做交互

Transformations

  • map(func)
    返回一个新的分布式数据集,由每个原元素经过func函数转换后组成

  • flatMap(func)
    类似于map,但是每一个输入元素,会被映射为0到多个输出元素(因此,func函数的返回值是一个Seq,而不是单一元素)

  • map VS. flatMap

    //map
    scala> val input = sc.parallelize(List(1,2,3,4))
    input: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at parallelize at <console>:21
    scala> val result = input.map(x => x*x*x)
    result: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[7] at map at <console>:23
    scala> result.collect()
    res16: Array[Int] = Array(1, 8, 27, 64)
    
    //flatMap
    scala> val flat = input.flatMap(x => List(x,x*x,x*x*x))
    flat: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[9] at flatMap at <console>:23
    scala> flat.collect()
    res19: Array[Int] = Array(1, 1, 1, 2, 4, 8, 3, 9, 27, 4, 16, 64)
    
  • filter(func)
    返回一个新的数据集,由经过func函数后返回值为true的原元素组成

  • sample(withReplacement, frac, seed)
    根据给定的随机种子seed,随机抽样出数量为frac的数据

  • union(otherDataset)
    返回一个新的数据集,由原数据集和参数联合而成

  • intersection(otherRDD)
    返回一个新的数据集,由两个数据集都有的元素构成

  • subtract(otherRDD)
    返回一个新的数据集,数据由原数据集减去otherRDD中的数据组成

  • distinct()
    返回一个新的数据集,数据中的元素唯一

  • 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)对的数据集,所有元素交互进行笛卡尔积。