spark RDD Operations

RDD支持两种类型的操作： transformations-转换 ，将一个已经存在的RDD转换为一个新的RDD，另外一种称为 actions-动作 ，动作算子一般在执行结束以后，会将结果返回给Driver。在Spark中所有transformations 都是lazy的，所有转换算子并不会立即执行，它们仅仅是记录对当前RDD的转换逻辑。仅当 Actions 算子要求将结果返回给Driver程序时 transformations 才开始真正的进行转换计算。这种设计使Spark可以更高效地运行。

默认情况下，每次在其上执行操作时，都可能会重新计算每个转换后的RDD。但是，您也可以使用persist（或cache）方法将RDD保留在内存中，在这种情况下，Spark会将元素保留在群集中，以便下次查询时可以更快地进行访问。

scala> var rdd1=sc.textFile("hdfs:///demo/words/t_word",1).map(line=>line.split("
").length)
rdd1: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[117] at map at <console>:24
scala> rdd1.cache
res54: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[117] at map at <console>:24
scala> rdd1.reduce(_+_)
res55: Int = 15
scala> rdd1.reduce(_+_)
res56: Int = 15

Spark还支持将RDD持久存储在磁盘上，或在多个节点之间复制。比如用户可调用 persist(StorageLevel.DISK_ONLY_2) 将RDD存储在磁盘上，并且存储2份。

Transformations

√map( func )

将一个RDD[U] 转换为 RRD[T]类型。在转换的时候需要用户提供一个匿名函数 func: U => T

scala> var rdd:RDD[String]=sc.makeRDD(List("a","b","c","a"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[120] at makeRDD at
<console>:25
scala> val mapRDD:RDD[(String,Int)] = rdd.map(w => (w, 1))
mapRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[121] at map at
<console>:26

√filter( func )

将对一个RDD[U]类型元素进行过滤，过滤产生新的RDD[U],但是需要用户提供 func:U => Boolean
系统仅会保留返回true的元素。

scala> var rdd:RDD[Int]=sc.makeRDD(List(1,2,3,4,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[122] at makeRDD at
<console>:25
scala> val mapRDD:RDD[Int]=rdd.filter(num=> num %2 == 0)
mapRDD: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[123] at filter at
<console>:26
scala> mapRDD.collect
res63: Array[Int] = Array(2, 4)

√flatMap( func )

和map类似，也是将一个RDD[U] 转换为 RRD[T]类型。但是需要用户提供一个方法 func:U => Seq[T]

scala> var rdd:RDD[String]=sc.makeRDD(List("this is","good good"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[124] at makeRDD at
<console>:25
scala> var flatMapRDD:RDD[(String,Int)]=rdd.flatMap(line=> for(i<- line.split("\\s+"))
yield (i,1))
flatMapRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[125] at flatMap
at <console>:26
scala> var flatMapRDD:RDD[(String,Int)]=rdd.flatMap( line=>
line.split("\\s+").map((_,1)))
flatMapRDD: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[126] at flatMap
at <console>:26
scala> flatMapRDD.collect
res64: Array[(String, Int)] = Array((this,1), (is,1), (good,1), (good,1))

√mapPartitions( func )

和mapPartitions类似，但是该方法会提供RDD元素所在的分区编号。因此 func:(Int, Iterator)=>Iterator

scala> var rdd:RDD[Int]=sc.makeRDD(List(1,2,3,4,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[128] at makeRDD at
<console>:25
scala> var mapPartitionsRDD=rdd.mapPartitions(values => values.map(n=>(n,n%2==0)))
mapPartitionsRDD: org.apache.spark.rdd.RDD[(Int, Boolean)] = MapPartitionsRDD[129] at
mapPartitions at <console>:26
scala> mapPartitionsRDD.collect
res70: Array[(Int, Boolean)] = Array((1,false), (2,true), (3,false), (4,true),
(5,false))

√mapPartitionsWithIndex

和mapPartitions类似，但是该方法会提供RDD元素所在的分区编号。因此 func:(Int, Iterator)=> Iterator

scala> var rdd:RDD[Int]=sc.makeRDD(List(1,2,3,4,5,6),2)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[139] at makeRDD at
<console>:25
scala> var mapPartitionsWithIndexRDD=rdd.mapPartitionsWithIndex((p,values) =>
values.map(n=>(n,p)))
mapPartitionsWithIndexRDD: org.apache.spark.rdd.RDD[(Int, Int)] =
MapPartitionsRDD[140] at mapPartitionsWithIndex at <console>:26
scala> mapPartitionsWithIndexRDD.collect
res77: Array[(Int, Int)] = Array((1,0), (2,0), (3,0), (4,1), (5,1), (6,1))

sample

sample( withReplacement , fraction , seed ) 抽取RDD中的样本数据，可以通过 withReplacement ：是否允许重复抽样、 fraction :控制抽样大致比例、 seed :控制的是随机抽样过程中产生随机数。

scala> var rdd:RDD[Int]=sc.makeRDD(List(1,2,3,4,5,6))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[150] at makeRDD at
<console>:25
scala> var simpleRDD:RDD[Int]=rdd.sample(false,0.5d,1L)
simpleRDD: org.apache.spark.rdd.RDD[Int] = PartitionwiseSampledRDD[151] at sample at
<console>:26
scala> simpleRDD.collect
res91: Array[Int] = Array(1, 5, 6)

种子不一样，会影响最终的抽样结果！

union( otherDataset )

是将两个同种类型的RDD的元素进行合并。

scala> var rdd:RDD[Int]=sc.makeRDD(List(1,2,3,4,5,6))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[154] at makeRDD at
<console>:25
scala> var rdd2:RDD[Int]=sc.makeRDD(List(6,7))
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[155] at makeRDD at
<console>:25
scala> rdd.union(rdd2).collect
res95: Array[Int] = Array(1, 2, 3, 4, 5, 6, 6, 7)

intersection( otherDataset )

是将两个同种类型的RDD的元素进行计算交集。

scala> var rdd:RDD[Int]=sc.makeRDD(List(1,2,3,4,5,6))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[154] at makeRDD at
<console>:25
scala> var rdd2:RDD[Int]=sc.makeRDD(List(6,7))
rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[155] at makeRDD at
<console>:25
scala> rdd.intersection(rdd2).collect
res100: Array[Int] = Array(6)

distinct([ numPartitions ]))

去除RDD中重复元素,其中 numPartitions是一个可选参数，是否修改RDD的分区数，一般是在当数据集经过去重之后，如果数据量级大规模降低，可以尝试传递 numPartitions 减少分区数。

scala> var rdd:RDD[Int]=sc.makeRDD(List(1,2,3,4,5,6,5))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[154] at makeRDD at
<console>:25
scala> rdd.distinct(3).collect
res106: Array[Int] = Array(6, 3, 4, 1, 5, 2)

√join

√join( otherDataset , [ numPartitions ])

当调用RDD[(K,V)]和RDD[(K,W)]系统可以返回一个新的RDD[(k,(v,w))]（默认内连接）,目前支leftOuterJoin, rightOuterJoin, 和 fullOuterJoin.

scala> var userRDD:RDD[(Int,String)]=sc.makeRDD(List((1,"zhangsan"),(2,"lisi")))
userRDD: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[204] at
makeRDD at <console>:25
scala> case class OrderItem(name:String,price:Double,count:Int)
defined class OrderItem
scala> var
orderItemRDD:RDD[(Int,OrderItem)]=sc.makeRDD(List((1,OrderItem("apple",4.5,2))))
orderItemRDD: org.apache.spark.rdd.RDD[(Int, OrderItem)] = ParallelCollectionRDD[206]
at makeRDD at <console>:27
scala> userRDD.join(orderItemRDD).collect
res107: Array[(Int, (String, OrderItem))] = Array((1,
(zhangsan,OrderItem(apple,4.5,2))))
scala> userRDD.leftOuterJoin(orderItemRDD).collect
res108: Array[(Int, (String, Option[OrderItem]))] = Array((1,
(zhangsan,Some(OrderItem(apple,4.5,2)))), (2,(lisi,None)))

cartesian了解

cartesian( otherDataset )-了解

计算集合笛卡尔积

scala> var rdd1:RDD[Int]=sc.makeRDD(List(1,2,4))
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[238] at makeRDD at
<console>:25
scala> var rdd2:RDD[String]=sc.makeRDD(List("a","b","c"))
rdd2: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[239] at makeRDD at
<console>:25
scala> rdd1.cartesian(rdd2).collect
res120: Array[(Int, String)] = Array((1,a), (1,b), (1,c), (2,a), (2,b), (2,c), (4,a),
(4,b), (4,c))

coalesce( numPartitions )

当经过大规模的过滤数据以后，可以使 coalesce 对RDD进行分区的缩放（只能减少分区，不可以增加）。

scala> var rdd1:RDD[Int]=sc.makeRDD(0 to 100)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[252] at makeRDD at
<console>:25
scala> rdd1.getNumPartitions
res129: Int = 6
scala> rdd1.filter(n=> n%2 == 0).coalesce(3).getNumPartitions
res127: Int = 3
scala> rdd1.filter(n=> n%2 == 0).coalesce(12).getNumPartitions
res128: Int = 6

在这里插入代码片

repartition( numPartitions )

和 coalesce 相似，但是该算子能够变大或者缩小RDD的分区数。

scala> var rdd1:RDD[Int]=sc.makeRDD(0 to 100)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[252] at makeRDD at
<console>:25
scala> rdd1.getNumPartitions
res129: Int = 6
scala> rdd1.filter(n=> n%2 == 0).repartition(12).getNumPartitions
res130: Int = 12
scala> rdd1.filter(n=> n%2 == 0).repartition(3).getNumPartitions
res131: Int = 3

√xxxByKey-算子(掌握)

在Spark中专门针对RDD[(K,V)]类型数据集提供了xxxByKey算子实现对RDD[(K,V)]类型针对性实现计算。

groupByKey([ numPartitions ])

类似于MapReduce计算模型。将RDD[(K, V)] 转换为RDD[ (K, Iterable)]

scala> var lines=sc.parallelize(List("this is good good"))
lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at
<console>:24
scala> lines.flatMap(_.split("\\s+")).map((_,1)).groupByKey.collect
res3: Array[(String, Iterable[Int])] = Array((this,CompactBuffer(1)),
(is,CompactBuff)), (good,CompactBuffer(1, 1)))

groupBy(f:(k,v)=> T)

scala> var lines=sc.parallelize(List("this is good good"))
lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at
<console>:24
scala> lines.flatMap(_.split("\\s+")).map((_,1)).groupBy(t=>t._1)
res5: org.apache.spark.rdd.RDD[(String, Iterable[(String, Int)])] = ShuffledRDD[18] at
groupBy at <console>:26
scala> lines.flatMap(_.split("\\s+")).map((_,1)).groupBy(t=>t._1).map(t=>
(t._1,t._2.size)).collect
res6: Array[(String, Int)] = Array((this,1), (is,1), (good,2))

reduceByKey

reduceByKey( func , [ numPartitions ])

scala> var lines=sc.parallelize(List("this is good good"))
lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at
<console>:24scala> lines.flatMap(_.split("\\s+")).map((_,1)).collect
res21: Array[(String, Int)] = Array((this,1), (is,1), (good,1), (good,1))scala> lines.flatMap(_.split("\\s+")).map((_,1)).reduceByKey(_+_).collect
res8: Array[(String, Int)] = Array((this,1), (is,1), (good,2))

aggregateByKey

aggregateByKey( zeroValue )( seqOp , combOp , [ numPartitions ])

在（K，V）对的数据集上调用时，返回（K，U）对的数据集，其中每个键的值使用给定的组合函数和中性“零”值进行聚合。允许聚合值类型，即di！与输入值类型不同，同时避免不必要的分配。就像在groupByKey，reduce任务的数量可以通过可选的第二个参数配置。

scala> var lines=sc.parallelize(List("this is good good"))
lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at
<console>:24
scala> lines.flatMap(_.split("\\s+")).map((_,1)).aggregateByKey(0)(_+_,_+_).collect
res9: Array[(String, Int)] = Array((this,1), (is,1), (good,2))

sortByKey

sortByKey([ ascending ], [ numPartitions ])

在K实现有序的（K，V）对数据集上调用时，返回（K，V）对数据集按键升序或降序排序，如布尔升序参数中指定的 sortByKey(true/false)默认按照第一个字段 排序 true升序

scala> var lines=sc.parallelize(List("this is good good"))
lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at
<console>:24
scala> lines.flatMap(_.split("\\s+")).map((_,1)).aggregateByKey(0)
(_+_,_+_).sortByKey(true).collect
res13: Array[(String, Int)] = Array((good,2), (is,1), (this,1))
scala> lines.flatMap(_.split("\\s+")).map((_,1)).aggregateByKey(0)
(_+_,_+_).sortByKey(false).collect
res14: Array[(String, Int)] = Array((this,1), (is,1), (good,2))

sortBy

sortBy(字段，true/false) 第一个参数代表用第几个字段排序 第二个参数代表升序/降序

scala> var lines=sc.parallelize(List("this is good good"))
lines: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[0] at parallelize at
<console>:24
scala> lines.flatMap(_.split("\\s+")).map((_,1)).aggregateByKey(0)
(_+_,_+_).sortBy(_._2,false).collect
res18: Array[(String, Int)] = Array((good,2), (this,1), (is,1))
scala> lines.flatMap(_.split("\\s+")).map((_,1)).aggregateByKey(0)
(_+_,_+_).sortBy(t=>t,false).collect
res19: Array[(String, Int)] = Array((this,1), (is,1), (good,2))

Actions

Spark任何一个计算任务，有且仅有一个动作算子，用于触发job的执行。将RDD中的数据写出到外围系统或者传递给Driver主程序。

reduce( func )

该算子能够对远程结果进行计算，然后将计算结果返回给Driver。计算文件中的字符数。

scala> sc.textFile("file:///root/t_word").map(_.length).reduce(_+_)
res3: Int = 64

collect()

将远程RDD中数据传输给Driver端。通常用于测试环境或者RDD中数据非常的小的情况才可以使用Collec算子，否则Driver可能因为数据太大导致内存溢出。

scala> sc.textFile("file:///root/t_word").collect
res4: Array[String] = Array(this is a demo, hello spark, "good good study ", "day day
up ", come on baby)

foreach( func )

在数据集的每个元素上运行函数func。通常这样做是出于副作用，例如更新累加器或与 外部存储系统 交互。

scala> sc.textFile("file:///root/t_word").foreach(line=>println(line))

count()

返回RDD中元素的个数

scala> sc.textFile("file:///root/t_word").count()
res7: Long = 5

first()|take( n )

first 得到第一个元素，take(n) 得到前n个元素 默认从0开始

scala> sc.textFile("file:///root/t_word").first
res9: String = this is a demo
scala> sc.textFile("file:///root/t_word").take(1)
res10: Array[String] = Array(this is a demo)
scala> sc.textFile("file:///root/t_word").take(2)
res11: Array[String] = Array(this is a demo, hello spark)

takeSample

takeSample( withReplacement , num , [ seed ])

随机的从RDD中采样num个元素，并且将采样的元素返回给Driver主程序。因此这和sample转换算子有很大的区别。第三个参数默认可以不给，如果给了 1L 那么1L得到的不会变， 2L同理

scala> sc.textFile("file:///root/t_word").takeSample(false,2)
res20: Array[String] = Array("good good study ", hello spark)scala>  sc.textFile("file:///root/t_word").takeSample(false,3,1L)
res55: Array[String] = Array(come on baby, good good study, this is a demo)scala>  sc.textFile("file:///root/t_word").takeSample(false,3,2L)
res56: Array[String] = Array(come on baby, this is a demo, hello spark)scala>  sc.textFile("file:///root/t_word").takeSample(false,3,1L)
res57: Array[String] = Array(come on baby, good good study, this is a demo)

takeOrdered

takeOrdered( n , [ordering] )

返回RDD中前N个元素，用户可以指定比较规则

scala> case class User(name:String,deptNo:Int,salary:Double)
defined class User
scala> var
userRDD=sc.parallelize(List(User("zs",1,1000.0),User("ls",2,1500.0),User("ww",2,1000.0
)))
userRDD: org.apache.spark.rdd.RDD[User] = ParallelCollectionRDD[51] at parallelize at
<console>:26
scala> userRDD.takeOrdered
def takeOrdered(num: Int)(implicit ord: Ordering[User]): Array[User]
scala> userRDD.takeOrdered(3)
<console>:26: error: No implicit Ordering defined for User.
userRDD.takeOrdered(3)
scala> implicit var userOrder=new Ordering[User]{
| override def compare(x: User, y: User): Int = {
| if(x.deptNo!=y.deptNo){
| x.deptNo.compareTo(y.deptNo)
| }else{
| x.salary.compareTo(y.salary) * -1
| }
| }
| }
userOrder: Ordering[User] = $anon$1@7066f4bc
scala> userRDD.takeOrdered(3)
res23: Array[User] = Array(User(zs,1,1000.0), User(ls,2,1500.0), User(ww,2,1000.0))

saveAsTextFile( path )

将数据集的元素作为文本文件（或一组文本文件）写入本地的给定目录中文件系统、HDFS或任何其他Hadoop支持的文件系统。Spark会调用RDD中元素的toString方法将元素以文本行的形式写入到文件中。

scala> sc.textFile("file:///root/t_word").flatMap(_.split("
")).map((_,1)).reduceByKey(_+_).sortBy(_._1,true,1).map(t=>
t._1+"\t"+t._2).saveAsTextFile("hdfs:///demo/results02")

saveAsSequenceFile( path )

该方法只能用于RDD[(k,v)]类型。并且K/v都必须实现Writable接口，由于使用Scala编程，Spark已经实现隐式转换将Int, Double, String, 等类型可以自动的转换为Writable

scala> sc.textFile("file:///root/t_word").flatMap(_.split("
")).map((_,1)).reduceByKey(_+_).sortBy(_._1,true,1).saveAsSequenceFile("hdfs:///demo/r
esults03")scala> sc.sequenceFile[String,Int]("hdfs:///demo/results03").collect
res29: Array[(String, Int)] = Array((a,1), (baby,1), (come,1), (day,2), (demo,1),
(good,2), (hello,1), (is,1), (on,1), (spark,1), (study,1), (this,1), (up,1))