spark RDD算子大全

map()&&flatMap()

map()&&mapPartitions()

mapPartitionsWithIndex()

filter()

take()&&takeOrdered()&&top()&&first()

sample()&&takeSample()

union()&&intersection()&&subtract()

reduce()&&fold()

fold()&&aggregate()

zipWithUniqueId()&&zipWithIndex()

zip()&&zipPartitions()

collect()&&toArray()

foreach()

count()

combineByKey()

reduceByKey()

groupBy()&&groupByKey()

ShuffledRDD类

map()&&flatMap()

返回一个新的RDD

    val data=sc.parallelize(0 to 4,2)val m=data.map(x=>x to 3)m.foreach(println)println(m.count())//RDD元素数量val fm=data.flatMap(x=>x to 3)fm.foreach(println)println(fm.count())

map结果：Range(0, 1, 2, 3)，Range(1, 2, 3)，Range(2, 3)，Range(3)，Range() RDD元素数量5

flatMap结果：0 1 2 3 1 2 3 2 3 3 RDD元素数量10

可以看到map()是将一个元素生成一个元素，而flatMap()是将一个元素生成N（也可能0个）个元素。

    val data=sc.parallelize(List("tom tom","jerry hehe","hehe"),2)val m=data.map(x=>x.split(" "))m.foreach(println)println(m.count())val fm=data.flatMap(x=>x.split(" "))fm.foreach(println)println(fm.count())

map结果：java.lang.String;@226d19fb，java.lang.String;@ffdbd08，java.lang.String;@3d58c9e5 元素数量3

flatMap结果：tom tom jerry hehe hehe 元素数量5

可以看到map将每个元素转为数组，新RDD的元素类型为数组，而faltMap是将所有数组中元素取出并连接起来。

map()&&mapPartitions()

返回一个新的RDD

map()与mapPartitions()源码：

class MappedRDD[U: ClassManifest, T: ClassManifest](prev: RDD[T],f: T => U)extends RDD[U](prev.context) {override def splits = prev.splitsoverride val dependencies = List(new OneToOneDependency(prev))override def compute(split: Split) = prev.iterator(split).map(f)//分区中每个元素执行f方法
}class MapPartitionsRDD[U: ClassManifest, T: ClassManifest](prev: RDD[T],f: Iterator[T] => Iterator[U])extends RDD[U](prev.context) {override def splits = prev.splitsoverride val dependencies = List(new OneToOneDependency(prev))override def compute(split: Split) = f(prev.iterator(split))//每个分区的数据作为一个整体执行f
}

下面看示例：

    val data = sc.parallelize(1 to 10,3)data.mapPartitions(x=>mapParrtitionsF(x)).count()data.map(x=>mapF(x)).count()def mapF(x:Int):Int={println("调用了mapF方法")x}def mapParrtitionsF(x:Iterator[Int]):Iterator[Int]={println("调用了mapParrtitionsF方法")x}

map结果：调了10次方法，RDD有多少元素执行多少次方法

flatMap结果：调了3次方法，RDD有多少分区执行多少次方法

mapPartitionsWithIndex()

返回一个新的RDD

    val data = sc.parallelize(1 to 5,3)data.mapPartitionsWithIndex((x,iter)=>{var result=List[String]()while (iter.hasNext){result ::=(x+"-"+iter.next())}result.toIterator}).foreach(println)

结果为：0-1，1-2，1-3，2-4，2-5

mapPartitionsWithIndex()传入的方法需要两个参数，一个为分区Id，另一个为分区数据，该方法可用来查看各分区的数据

filter()

返回一个新的RDD

    val data = sc.parallelize(1 to 5,3)data.filter(x=>x%2==0).foreach(println)

结果为: 2 4

filter()接收一个返回值为布尔类型的方法，过滤掉不符合条件的元素，符合条件的元素组成一个新的RDD

take()&&takeOrdered()&&top()&&first()

前三个返回一个数组，最后一个返回一个值

   def take(num: Int): Array[T] = {if (num == 0) {return new Array[T](0)}val buf = new ArrayBuffer[T]//存放取出的元素的数组var p = 0while (buf.size < num && p < splits.size) {//循环，直到数取够或者循环完所有分区val left = num - buf.size//得到还需要从取出的元素数//true为允许job本节点执行，无需发往集群val res = sc.runJob(this, (it: Iterator[T]) => it.take(left).toArray, Array(p), true)buf ++= res(0)//将从第一个分区取到的数据存入数组if (buf.size == num)//若已取够，返回，无的话继续取后面的分区return buf.toArrayp += 1}return buf.toArray}

该方法为取出RDD的n个元素，先从一个分区中取数据，取够的话，直接返回，没够的话，继续取别的分区。

    val data=sc.parallelize(List(6,3,2,0,11,45,1),3)for(e<-data.take(2)){println(e)}for(e<-data.takeOrdered(2)){println(e)}for(e<-data.top(2)){println(e)}println(data.first())

take(2)结果为：6 3 ，取前两个元素

takeOrdered(2)结果为：0 1 ，升序排列后，取前两元素

top(2)结果为:45 11，降序排列后，取前两元素

first()结果为:6 取第一个元素

sample()&&takeSample()

一个返回新的RDD 一个返回数组

    val data = sc.parallelize(1 to 10000,3)println(data.sample(true,0.5).count())println(data.takeSample(true,5000).length)

sample()结果为:4976

takeSample()结果为:5000

这两算子都是随机抽取元素，可以传三个参数，第一个为布尔值，表示是否重复抽取元素，第二个sample()为抽取比例，takeSample()为具体数值，第三个为随机种子，java的随机数是伪随机数，通过计算随机种子得到某些数，可用默认值。

从结果可以看出，抽取比例为0.5时是抽不到5000个元素的，如果想抽取具体多少元素，可用takeSample()

union()&&intersection()&&subtract()

返回一个新的RDD

    val d4=sc.parallelize(List(1,3,4,6,3,8,1),2)val d5=sc.parallelize(List(2,3,4,5,3,8,9),3)d4.union(d5).foreach(println)println(d4.union(d5).partitions.length)//新RDD分区数d4.intersection(d5).foreach(println)println(d4.intersection(d5).partitions.length)println(d4.intersection(d5,6).partitions.length)//指定新RDD分区数d4.subtract(d5).foreach(println)println(d5.subtract(d4).partitions.length)println(d4.subtract(d5).partitions.length)

union()结果为：1,3,4,6,3,8,1,2,3,4,5,3,8,9 分区数为2+3=5

intersection()结果为：8,3,4 分区数为：3 6

subtract()结果为：1,6,1 分区数为：3 2

这三个算子分别求：并集，交集，差集。其中intersection()交集去重，别的union()，subtract()均不去重，union()分区数为两个RDD分区数之和，intersection()分区默认为max(rdd1分区，rdd2分区)，可以手动指定分区数，subtract()默认分区数为调该方法的RDD的分区数，也可以指定。

reduce()&&fold()

返回一个值

    def reduce(f: (T, T) => T): T = {val cleanF = sc.clean(f)//如果迭代器有值，调用reduceLeft()val reducePartition: Iterator[T] => Option[T] = iter => {if (iter.hasNext) {Some(iter.reduceLeft(cleanF))}else {None}}//提交job,runJob返回数组val options = sc.runJob(this, reducePartition)val results = new ArrayBuffer[T]//遍历返回的结果，存入数组中for (opt <- options; elem <- opt) {results += elem}if (results.size == 0) {throw new UnsupportedOperationException("empty collection")} else {//数组继续调用reduceLeft()方法，将结果返回return results.reduceLeft(cleanF)}}

我们知道，提交task的时候，一个RDD的分区数据对应一个迭代器，一个迭代器对应一个task，

所以，runJob()返回来的数组存放的是各分区的结果。

reduce()主要分两步，第一步调用reduceLeft()计算各个分区数据，第二步，再调用reduceLeft()计算各分区结果，返回。

接下来我们说下Iterator的reduceLeft()方法

    val w=(1 to 5).toIteratorprintln(w.reduceLeft(_+_))  //15println(w.reduceLeft(_-_))  //-13println(w.reduceRight(_-_)) //3

reduceLeft()计算的是从左边开始相邻两个元素的值，将结果与后面元素再计算，直到所有元素参与计算

第一个输出:(((1+2)+3)+4)+5=15

第二个输出:(((1-2)-3)-4)-5=-13

第三个输出:从右边开始,1-(2-(3-(4-5)))=3

  def fold(zeroValue: T)(op: (T, T) => T): T = {val cleanOp = sc.clean(op)val results = sc.runJob(this, (iter: Iterator[T]) => iter.fold(zeroValue)(cleanOp))return results.fold(zeroValue)(cleanOp)}

fold跟reduce是差不多，只不过fold有初值，先计算各个分区与初值的结果，存入数组，再计算结果与初值的值。

可以得出当有1个分区时,初值被计算2次，第一次与分区数据，第二次与分区结果数据

val a=sc.parallelize(List(1,2,3,4,5),1)
val b=sc.parallelize(List(1,2,3,4,5),2)
val c=sc.parallelize(List(1,2,3,4,5),3)println(a.fold(10)(_+_))
println(b.fold(10)(_+_))
println(c.fold(10)(_+_))

a的结果为:1+2+3+4+5+10=25, 25+10=35

b的结果为:1+2+3+10=16，4+5+10=19，16+19+10=45 （分区数据编的，与真实情况可能有区别）

c的结果为:1+2+10=13，3+4+10=17，5+10=15，13+17+15=55

    val data = sc.parallelize(1 to 5,3)println(data.reduce(_+_))println(data.fold(5)(_+_))

reduce()结果为：15

fold()结果为：35

关于reduce()与fold()源码可看我之前的博客。

reduce()的计算过程:(((1+2)+3)+4)+5，而fold()的计算过程是先计算各分区与初值的结果，最后计算各分区结果与初值，

假设data的数据分布是:(1)，（2，3），（4，5）

则fold()的计算结果是:(1+5)+(2+3+5)+(4+5+5)+5，

所以reduce()与fold()的区别就是一个有初值，一个无初值。

fold()&&aggregate()

返回一个值

   def aggregate[U: ClassManifest](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U):     U = {val cleanSeqOp = sc.clean(seqOp)val cleanCombOp = sc.clean(combOp)val results = sc.runJob(this,(iter: Iterator[T]) => iter.aggregate(zeroValue)(cleanSeqOp, cleanCombOp))return results.fold(zeroValue)(cleanCombOp)}

aggregate()先用seqOp方法计算各分区里的数据，将各分区结果存入数组，然后再调用fold()，方法为combOp计算结果数据。

可以看到aggregate()与fold()的区别在于：flod计算某个分区数据与最后计算各分区结果数据用的用同一个方法，而aggregate()

则可以用两种不同的方法，当aggregate()两个方法一样时，结果与fold()是一样的。

    val a=sc.parallelize(List(1,2,3,4),2)println(a.aggregate(5)(_+_,_+_))println(a.fold(5)(_+_))

结果都为 25，计算步骤可参考上面的fold()方法

    val data = sc.parallelize(1 to 5,3)println(data.fold(5)(_+_))println(data.aggregate(5)(_+_,_*_))

fold()结果为：35

aggregate()结果为：4200

aggregate()需要传入两个方法，第一个方法计算各分区内数据，第二个方法计算各分区之前结果数据。

aggregate()计算过程:(1+5)*(2+3+5)*(4+5+5)*5=4200

而fold()传一个方法，计算各分区内数据不仅用这个方法，计算各分区之前结果数据也用该方法。

所以fold()与aggregate()区别是：一个传一个方法，一个传两个方法。

zipWithUniqueId()&&zipWithIndex()

返回一个新的RDD

    val data = sc.parallelize(1 to 5,3)data.zipWithUniqueId().foreach(println)data.zipWithIndex().foreach(println)

zipWithUniqueId()结果为：(2，1)，（3，4），（4，2），（5，5），（1，0）

zipWithIndex()结果为：（4，3），（2，1），（3，2），（5，4），（1，0）

可以看出，这两种方法都是返回一个键值对的RDD，键为元素，zipWithIndex()值为下标，从0到RDD元素数-1，而zipWithUniqueId()值为唯一的Id，不受最大值为RDD元素数-1的约束

zip()&&zipPartitions()

返回一个新的RDD

    val d1=sc.parallelize(List("tom","jerry","hehe","zlq","wnn","nm","sl"),2)val d3=sc.parallelize(List(6,3,2,0,11,45,1),2)d1.zip(d3).foreach(println)d1.zipPartitions(d3)((iter1,iter2)=>{//方法的参数为迭代器var result=List[String]()while (iter1.hasNext&&iter2.hasNext){result ::=(iter1.next()+"-"+iter2.next())}result.iterator}).foreach(println)

zip()结果为：(zlq,0)，(tom,6)，(wnn,11)，(nm,45)，(sl,1)，(jerry,3)，(hehe,2)

zipPartitions()结果为：hehe-2，jerry-3，tom-6，sl-1，nm-45，wnn-11，zlq-0

zip()要求两个RDD分区数与元素数必须相等，而zipPartitions()只要求分区数相同。生成新RDD与原RDD分区数相同

collect()&&toArray()

collect()方法将RDD中的数据转化为数组

def collect(): Array[T] = {val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)//将各分区数据连接起来Array.concat(results: _*)}

def toArray(): Array[T] = collect()

toArray方法，调用collect()方法，将RDD转为数组

    val a=sc.parallelize(1 to 5,2)println(a.collect()(1))//打印下标为1的值

将打印 2

foreach()

def foreach(f: T => Unit) {val cleanF = sc.clean(f)//提交job，迭代器每个元素执行传入方法sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))}

iter.foreach是无返回值，一般用于调试打印RDD中的数据或将RDD的数据输出到别的文件系统

    val a=sc.parallelize(1 to 5,2)a.foreach(print)

将打印12345

count()

  def count(): Long = {sc.runJob(this, (iter: Iterator[T]) => {var result = 0Lwhile (iter.hasNext) {//遍历迭代器中所有元素result += 1Liter.next}result}).sum}

count()用来求RDD中元素的个数，求出个分区元素个数，然后sum()求和

    val w=sc.parallelize(0 to 10,3).count()println(w)

结果11

combineByKey()

//参数：
//创建聚合器，如果K已经创建，则调mergeValue，没创建，则创建，将V生成一个新值
//作用于分区内的数据，将相同K对应的V聚合
//作用于各分区，将各分区K相同的V聚合
def combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C) : RDD[(K, C)] = {combineByKey(createCombiner, mergeValue, mergeCombiners, defaultPartitioner(self))}

//numSplits分区数，聚合完成后生成的RDD有几个分区
def combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C,numSplits: Int): RDD[(K, C)] = {combineByKey(createCombiner, mergeValue, mergeCombiners, new HashPartitioner(numSplits))}

def combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C,partitioner: Partitioner): RDD[(K, C)] = {val aggregator = new Aggregator[K, V, C](createCombiner, mergeValue, mergeCombiners)new ShuffledRDD(self, aggregator, partitioner)}

可以看到combineByKey()有三个重载的方法，最终都会调用第三个然后创建一个ShuffledRDD对象，

Partitioner默认按HashPartitioner分区

mergeValue与mergeCombiners两个方法与aggregate传的两个方法挺类似的，一个是处理某个分区内数据，一个处理所有分区的结果数据，spark的聚合算子大都调用combineByKey()算子，所以我们先去看别的算子createCombiner，mergeValue，mergeCombiners如何定义，更方便理解

reduceByKey()

def reduceByKey(func: (V, V) => V): RDD[(K, V)] = {reduceByKey(defaultPartitioner(self), func)}

def reduceByKey(func: (V, V) => V, numSplits: Int): RDD[(K, V)] = {//新RDD分区数将改变reduceByKey(new HashPartitioner(numSplits), func)}

def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)] = {combineByKey[V]((v: V) => v, func, func, partitioner)}

可以看到reduceByKey()有3个重载方法，一个传方法，一个传方法+新RDD分区数，一个传Partitioner+方法，最终调用

combineByKey()，mergeValue，mergeCombiners都为传入的方法。

    val w=sc.parallelize(List(("tom",1),("tom",3),("jerry",1),("jerry",7),("tom",5)),3)val s=w.reduceByKey(_*_)//将key相同的value相乘println(s.partitions.length)//打印新RDD分区数s.foreach(println)//打印新RDD元素

打印结果为：3 (jerry,7)(tom,15)

计算过程:假设3个分区数据为：

分区1 ("tom",1),("tom",3) 给tom创建聚合器(tom,1),因tom已经创建过,直接（tom，1*3）该分区结果为(tom,3)

分区2 ("jerry",1),("jerry",7) 同上，该分区结果为(jerry,7)

分区3 ("tom",5) 该分区结果为(tom,5)

最后再聚合各分区结果,(tom,3)，(jerry,7)，(tom,5) 最终结果为:(tom,3*5)，(jerry,7)

接下来我们用combineByKey()实现一下:

   val w=sc.parallelize(List(("tom",1),("tom",3),("jerry",1),("jerry",7),("tom",5)),3)w.combineByKey(x => (1, x),(c1: (Int,Int), y) => (c1._1 + 1, c1._2*y),//分区内数据计算方式(c1: (Int,Int), c2: (Int,Int)) => (c1._1 + c2._1, c1._2 * c2._2)//各分区结果之间计算方式).foreach(println)

打印结果为:(jerry,(2,7))(tom,(3,15))

combineByKey()与reduceByKey()的区别类似 fold()与aggregate()的区别，fold()与reduceByKey()是传一个方法，不仅作用于分区内数据，也作用于分区间结果数据，而aggregate()与combineByKey()是传两个方法，一个作用于分区内数据，另一个作用于分区间结果数据。

groupBy()&&groupByKey()

groupBy()

def groupBy[K: ClassManifest](f: T => K): RDD[(K, Seq[T])] = groupBy[K](f, sc.defaultParallelism)

//参数为，方法+新RDD分区数
def groupBy[K: ClassManifest](f: T => K, numSplits: Int): RDD[(K, Seq[T])] = {val cleanF = sc.clean(f)this.map(t => (cleanF(t), t)).groupByKey(numSplits)}

可以看到groupBy()是每个元素执行完方法后的结果作为key,元素作为value，再调用groupByKey()方法

    val w=sc.parallelize(0 to 9,3)w.groupBy(x=>x%3).foreach(println)

打印结果为:(0,CompactBuffer(0, 3, 6, 9)) (1,CompactBuffer(1, 4, 7)) (2,CompactBuffer(2, 5, 8))

元素除3余数相同的会放在同一个集合。

groupByKey()

  def groupByKey(): RDD[(K, Seq[V])] = {groupByKey(defaultPartitioner(self))}

  def groupByKey(numSplits: Int): RDD[(K, Seq[V])] = {groupByKey(new HashPartitioner(numSplits))}

  def groupByKey(partitioner: Partitioner): RDD[(K, Seq[V])] = {def createCombiner(v: V) = ArrayBuffer(v)，def mergeValue(buf: ArrayBuffer[V], v: V) = buf += v//放入变长数组中def mergeCombiners(b1: ArrayBuffer[V], b2: ArrayBuffer[V]) = b1 ++= b2//拼接两个数组val bufs = combineByKey[ArrayBuffer[V]](createCombiner _, mergeValue _, mergeCombiners _, partitioner)bufs.asInstanceOf[RDD[(K, Seq[V])]]}

可以看到最终调用的是combineByKey()方法，每个分区先将Key相同的value放入数组，最后再将各分区key相同的合并一个数组。

    val w=sc.parallelize(List(("tom",1),("tom",3),("jerry",1),("jerry",7),("tom",5)),2)w.groupByKey().foreach(println)println(w.groupByKey(5).partitions.length)//改变分区数

打印结果为:(tom,CompactBuffer(1, 3, 5)) (jerry,CompactBuffer(1, 7)) 5(分区数)

现在回过头看combineByKey()

  def combineByKey[C](createCombiner: V => C,mergeValue: (C, V) => C,mergeCombiners: (C, C) => C,partitioner: Partitioner): RDD[(K, C)] = {val aggregator = new Aggregator[K, V, C](createCombiner, mergeValue, mergeCombiners)new ShuffledRDD(self, aggregator, partitioner)}

创建了Aggregator对象，然后创建了ShuffledRDD对象。

ShuffledRDD类

class ShuffledRDD[K, V, C](parent: RDD[(K, V)],aggregator: Aggregator[K, V, C],part : Partitioner) extends RDD[(K, C)](parent.context) {//override val partitioner = Some(part)override val partitioner = Some(part)@transientval splits_ = Array.tabulate[Split](part.numPartitions)(i => new ShuffledRDDSplit(i))//长度为分区数的数组，存储各分区override def splits = splits_override def preferredLocations(split: Split) = Nil//创建shuffle依赖，注册shuffleIdval dep = new ShuffleDependency(context.newShuffleId, parent, aggregator, part)override val dependencies = List(dep)override def compute(split: Split): Iterator[(K, C)] = {val combiners = new JHashMap[K, C]//存储key,聚合值的mapdef mergePair(k: K, c: C) {val oldC = combiners.get(k)if (oldC == null) {//判断key之前是否遍历过combiners.put(k, c)//没的话，将k,v放入map中} else {//否则，将k对应的v,与之前的v的聚合值聚合combiners.put(k, aggregator.mergeCombiners(oldC, c))}}val fetcher = SparkEnv.get.shuffleFetcher//取数据fetcher.fetch[K, C](dep.shuffleId, split.index, mergePair)return new Iterator[(K, C)] {var iter = combiners.entrySet().iterator()def hasNext(): Boolean = iter.hasNext()def next(): (K, C) = {val entry = iter.next()(entry.getKey, entry.getValue)}}}
}

未完待续........