RDD的操作
1.1 概述
RDD整体包含两大类操作
transformation 从现有中创建一个新的数据集
action 在对数据集做一定程度的计算后将结果返回
对于所有的transformation,都是Lazy的,也就是说它不会立即执行,只是单纯的记住怎么样从原来的数据集进行转换的逻辑而已,它仅在某一个计算需要的情况下,才会被真正执行.
因为transformation 的Lazy性,RDD支持在每次计算时都进行重新计算,当然你可以将这个RDD保存下来 (persist or cache方法)避免每次重计算
可以通过设置不同的存储级别,将数据保存到硬盘,内存,或者选择同步多个副本到多个节点中.
1.2 集群环境下的变量与操作
集群环境,所有操作最终会交给executors去执行.而变量,会以数据副本的形式交给executors.很多时候,这与我们非集群环境下的开发思维有非常大的不同.
1.2.1 集群下的闭包
RDD是支持闭包操作的.但务必注意的是Spark不保证对闭包之外的对象引用进行的变化.
原因是闭包的会被序列化发生给每一个executor,对于闭包的之外的对象引用会拷贝一个副本给executor.这时多个executor执行至少是跨JVM的
这时对这个副本对象的变更没有任何意义,因为每个JVM(executor)的副本都是独立的.
1.2.2 集群下的print
集群环境下,print不会在driver端有任何输出.
原因也是一样,print最终是在每个executor执行,其输出也是在每个executor的stdout上,在driver端,是不会有这些输出的.
如果想在driver输出,一个比较简单的办法是调用collect()将结果发送到driver端在进行print,但这样可能会造成driver内存爆掉(所有executor的数据涌入).
比较推荐的做法是rdd.take(100).foreach(println)
1.2.3 共享变量
因为集群下,变量只会以副本方式交给executor,而不会将变量的值变化传回driver,所以一个可读写共享的变量是非常有用的.
Spark提供了两种共享变量 broadcast(广播变量) 和 accumulators(累加器)
1.2.3.1 广播变量(broadcast)
广播变量允许将一个只读变量的副本发送到每个机器上(executor机器),而不是对每一个任务发送一个副本.这样在同一机器上的多个任务,就可以反复使用这个变量了.
注意:
广播变量只会对每个节点分发一次,所以一般来说,广播变量不应该再被修改了.以保证每个广播变量的副本的值都是一致的
如果广播变量被修改,则需要将广播变量重新分发
另:
举个例子:Spark的action操作本身是通过一系列的stage来完成的,这些Stage是通过分布式的shuffle操作来进行切分的.而在每个Stage之间,Spark自动使用广播变量.
这里用法说明,只有数据本身会在多个Stage的多个任务中反复使用,或者说缓存这个数据是非常重要且非常必要的情况下,使用广播变量才有意义.
广播变量的使用如下:
// SparkContext.broadcast(v)进行创建,返回的是广播变量的封装器,以.value读取广播变量的的值 val broadcastVar = sc.broadcast(Array(1, 2, 3)) val v = broadcastVar.value
1.2.3.2 累加器(accumulators)
累加器变量仅支持累加操作,因为可以在并行计算执行一些特殊的计算(比计数或者求和).并且累加器的变化是可以在UI的Task界面上看见的(注,不支持Python)
累加器操作,依然遵循RDD的Lazy原则:
累加器更新操作是在Action中,并且在每个任务中只会执行一次(如果任务失败重启了,累加器更新不会执行)
而在transformation中,累加器依然不会立即执行更新,如果transformation被重新执行了,则累加器操作会重复执行
对于累加器变量,Spark原生支持数值类型.一个使用例子如下
val accum = sc.longAccumulator("My Accumulator")
sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))
println(accum.value)
也可以创建继承AccumulatorV2的类型,来实现自定义类型的累加器,例子如下:
//两个泛型参数->累加的元素类型和结果类型可以不同的 class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] { private val myVector: MyVector = MyVector.createZeroVector def reset(): Unit = { myVector.reset() } def add(v: MyVector): Unit = { myVector.add(v) } ... } // 创建一个自定义的累加器类型: val myVectorAcc = new VectorAccumulatorV2 //将这个触发器注册到SparkContext: sc.register(myVectorAcc, "MyVectorAcc1")
1.3 RDD的一些基本操作
1.3.1 Transformations 依赖关系
RDD是由父RDD经过转换函数形成一个个子RDD(子RDD依赖父RDD).针对不同的转换函数,以父RDD分区与子RDD分区的关系为标准,Spark将这些依赖关系分为两类.
窄依赖
窄依赖是指转换后,父RDD的每个分区只会被某一个子RDD分区使用.(一对一或者多对一的关系).
所以窄依赖一般出现在map,filter等子分区沿用父分区,不会发生重分区的时候.
宽依赖
宽依赖是指转换后,父RDD的某个或某些分区会被几个子RDD分区使用.(某个分区数据部分在这个RDD,部分在那个RDD,一对多关系)
宽依赖一般出现在groupByKey等子分区一定会发生重分区的时候
两种依赖关系的对比
一般来说,窄依赖比宽依赖对执行优化更加有利
i).窄依赖允许集群节点上以流水线的形式直接计算所有分区
宽依赖则需要先计算好父分区的分区信息,然后再以一个shuffle完成重分区,
ii).某个子分区异常需要重计算时,会对这个子分区所依赖的所有分区进行计算.(这是宽窄依赖都必须的),但是针对分区数据而言
窄依赖,一个或多个父分区完全对应一个子分区.对这些父分区的重计算,利用率是100%
宽依赖,父分区的数据不完全对应一个子分区(一对多关系,父分区的某些部分是其它分区的),但此时依然需要重计算父分区全部数据,造成计算浪费(因为白计算其它分区的数据)
1.3.2 Transformations 操作
map
对RDD中的元素执行一个指定函数,将执行结果作为新元素产生一个新的RDD.
与其它map系函数区别,map新元素的完全是Map函数的执行结果返回,所以新RDD的数量与老RDD是一一对应的.
val rdd = sc.parallelize(Seq("aa bb","cc dd","ee ff"),2)
rdd.map(rec=>rec.split(" ")>).collect().map(println(_))
//返回结果是rec.split(" ")结果(一维数组)=>[["aa","bb"],["cc","dd"]]
flatMap
与map相同,但结果会扁平化.即如果结果是迭代器容器的,会将元素从容器中取出再返回
val rdd = sc.parallelize(Seq("aa bb","cc"),2)
rdd.flatMap(rec=>rec.split(" ")).collect().map(println(_));
//返回结果["aa","bb","cc"]
//flatMap如以下这种方式使用是不行,flatMap返回结果必须是TraversableOnce[U](可迭代一次的类型)
//rdd.flatMap(rec=>(rec,1)).collect().map(println(_));
mapPartitions
与map相同,不过是以分区为单位,所以语法要求必须为 f: Iterator[T] => Iterator[U],注意返回结果不是以分区为单位,而是所有分区执行函数的结果的合并
val rdd = sc.parallelize(Seq("aa bb","cc dd","ee ff"),2)
rdd.mapPartitions(part=>part.map(rec=>rec.split(" "))).collect().map(println(_))
//结果是 [["aa","bb"],["cc","dd"],["ee","ff"]]
mapPartitionsWithIndex
与mapPartitions类似,不过它带有分区的index以供使用.所以语法要求为f: (Int, Iterator[T]) => Iterator[U]
val rdd = sc.parallelize(Seq("aa bb","cc dd","ee ff"),2)
rdd.mapPartitionsWithIndex((partIdx,part)=>part.map(rec=>(partIdx,rec))).collect().map(println(_))
//返回结果 (0,aa bb),(1,cc dd),(1,ee ff)
sample
抽样函数.可以从数据集中按一定比例抽取部分数据,抽取之后可以选择是否返回
语法要求 withReplacement: Boolean,fraction: Double,seed: Long = Utils.random.nextLong
val rdd = sc.parallelize(1 to 50) rdd.sample(false,0.2,System.currentTimeMillis).map(rec=>(rec,1)).collect().map(print(_)+" ") //返回结果 (8,1)(21,1)(26,1)(27,1)(34,1)(43,1)(46,1)(49,1)
union
将两个数据集合并(包含数据重复)
val rdd = sc.parallelize(1 to 10) val rdd2 = sc.parallelize(11 to 20) rdd.union(rdd2).map(rec=>rec.toString).collect().map(rec=>print(s"${rec} ")) //返回结果 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
intersection
将两个数据集合并,取交集作为结果返回
val rdd = sc.parallelize(1 to 10) val rdd2 = sc.parallelize(5 to 15) rdd.intersection(rdd2).map(rec=>rec.toString).collect().map(rec=>print(s"${rec} ")) //返回结果 6 7 9 8 10 5
distinct
对当前结果集去重返回
val rdd = sc.parallelize(1 to 10) val rdd2 = sc.parallelize(5 to 15) rdd.union(rdd2).distinct().map(rec=>rec.toString).collect().map(rec=>print(s"${rec} ")) //返回结果 4 14 6 8 12 10 2 13 15 11 1 3 7 9 5
groupByKey
将一个键值对类型的结果集按照key进行分组(如果是为分组聚合,groupByKey相比reduceByKey效率更低,因为少一个map-shuffer的combine)
val rdd = sc.parallelize(Seq("aa bb","cc dd","bb cc"),2)
rdd.flatMap(rec=>rec.split(" ")).map(rec=>(rec,1)).groupByKey().collect().map(rec=>print(s" ${rec._1} ${rec._2.sum} |"));
//返回结果 aa 1 | dd 1 | bb 2 | cc 2 |
reduceByKey
将一个键值对类型数据集,使用指定的函数分组聚合为另一个键值对类型数据集,(相比groupByKey性能更高,因为可以在map-shuffer进行combine减少数据量)
val rdd = sc.parallelize(Seq("aa bb","cc dd","bb cc"),2)
rdd.flatMap(rec=>rec.split(" ")).map(rec=>(rec,1)).reduceByKey((value1,value2)=>value1+value2).collect().map(rec=>print(s"${rec} "))
//返回结果 (aa,1) (dd,1) (bb,2) (cc,2)
aggregate
给出一个默认基准值,先使用seqOp遍历分区内元素传入基准值进行聚合,再对分区间结果使用combOp聚合为最后结果.
注意aggregate返回的结果直接是聚合结果(不是RDD),并且要与原RDD的类型一致
val rdd = sc.parallelize(1 to 10); /** * zeroValue:预定义一个初始值 (0,0) * seqOp: (U, T) => U 分区内元素聚合,遍历元素传入基准值执行函数.(类似Map-Shuffle) * U:当前基准值,T:当前元素 * 执行的逻辑是 (基准值(默认初始值), 元素No.1) 执行seqOp ,结果再作为基准值,执行(基准值(上步结果),元素No.2),以此类推 * combOp: (U, U) => U 分区间聚合,将各分区执行seqOp函数的结果再使用combOp聚合 (类似Reduce-Shuffle) */ val aggregateResult = rdd.aggregate((0,0))( seqOp=(sv,tv)=>(sv._1+tv,sv._2+1), combOp=(v1,v2)=>(v1._1+v2._1,v2._2+v2._2) ) println(aggregateResult) //输出结果 (55,10) (1-10的总和,1-10的个数) <=非RDD结果,并且类型必须是Int
aggregateByKey
与aggregate类似,但针对的是key分组,aggregateBykey是以key组为单位,对分组内元素遍历使用seqOp,再使用combOp聚合分组内
val rdd = sc.parallelize(Seq("a b c", "b c d"));
rdd.flatMap(rec => rec.split(" ")).map(rec => (rec, 1))
.aggregateByKey(0)(
seqOp = (sv, tv) => (sv + tv),
combOp = (v1, v2) => (v1 + v2)
)
.collect().map(rec => print(s"${rec} |"))
//输出结果 (d,1) |(b,2) |(a,1) |(c,2) |
sortByKey
将一个键值对RDD按key排序转换为另一个RDD
join
将两个键值对RDD((K, V),(K, W)),按Key合并为一个RDD(K, (V, W)) .(Spark同时还提供 leftOuterJoin,rightOuterJoin,fullOuterJoin)
val rdd = sc.parallelize(Seq("a b")).flatMap(rec => rec.split(" ")).map(rec => (rec, rec));
val rdd2 = sc.parallelize(Seq("b c")).flatMap(rec => rec.split(" ")).map(rec => (rec, rec));
rdd.join(rdd2).collect().map(rec => print(s"${rec} |"))
//两个RDD交集 (b,(b,b))
rdd.leftOuterJoin(rdd2).collect().map(rec => print(s"${rec} |"))
//leftOuterJoin左边全数据,右边Opt (b,(b,Some(b))) |(a,(a,None)) |
rdd.rightOuterJoin(rdd2).collect().map(rec => print(s"${rec} |"))
//rightOuterJoin右边全数据,左边Opt (b,(Some(b),b)) |(c,(None,c)) |
rdd.fullOuterJoin(rdd2).collect().map(rec => print(s"${rec} |"))
//笛卡尔乘积,Opt (b,(Some(b),Some(b))) |(a,(Some(a),None)) |(c,(None,Some(c))) |
cogroup
将多个键值对RDD按Key合并在一起.合并为全数据(没有丢失)
与fullOuterJoin区别在与多个RDD情况下,cogroup按key合并为一个,fullOuterJoin为多个的笛卡尔积
注意,如果某个数据集少某一个key,合并时是在这个数据集的位置上占CompactBuffer()的位置,而不是直接跳过
val rdd = sc.parallelize(Seq("a b")).flatMap(rec => rec.split(" ")).map(rec => (rec, rec));
val rdd2 = sc.parallelize(Seq("b c")).flatMap(rec => rec.split(" ")).map(rec => (rec, rec));
rdd.cogroup(rdd2).collect().map(rec => print(s"${rec} |"))
//(b,(CompactBuffer(b),CompactBuffer(b))) |(a,(CompactBuffer(a),CompactBuffer())) |(c,(CompactBuffer(),CompactBuffer(c))) |
cartesian
返回两个RDD的笛卡尔积结果
pipe
使用Shell的语法操作RDD
coalesce
重新调整RDD的分区后形成一个新的RDD.语法要求:numPartitions: Int, shuffle: Boolean = false.
numPartitions表示要重新调整的分区数量,shuffle表示重新调整分区时是否允许发生shuffle过程.
如果子分区数往下减少,则子分区数设置一定会成功.但要注意,在这种情况下会造成任务的并行度降低(分区数,任务数降了),任务内存更容易爆出(单个任务的数据增大了)
如果子分区数往上增加,则子分区数设置必须要设置shuffle=true,才会成功,否则子分区依然等于父分区
谨记:如果没有shuffle的参与,RDD只能减少分区(窄依赖),不能增加分区
repartition
只是coalesce的shuffle等于true的快捷方式. coalesce(numPartitions, shuffle = true)
repartitionAndSortWithinPartitions
1.3.2 Action 操作
reduce
RDD中元素前两个传给输入函数，产生一个新的return值，新产生的return值与RDD中下一个元素组成两个元素，再被传给输入函数，直到最后只有一个值为止
与reduceByKey的区别是 reduceByKey是一个转换操作,返回的是RDD, reduce是一个action操作,返回的是数据结果
val rdd = sc.parallelize(1 to 100,2); val value = rdd.reduce((v1,v2)=>v1+v2) println(value) //输出结果 5050
collect
将一个RDD的所有元素以数组的形式发回driver端.注意这个RDD必须是足够小的数据集,否则很容易将driver端的内存撑爆
count
返回一个RDD的元素的个数
first
返回一个RDD的第一个元素
take(n)
返回一个RDD的前N个元素
takeSample(withReplacement, num, [seed])
返回一个RDD的百分比抽样(withReplacement标识元素是否放回RDD以供多次使用)
takeOrdered(n, [ordering])
返回一个RDD按照设定的排序规则后的前N个元素
countByKey
只支持键值对类型,返回一个RDD的按照Key分组后的每组计数
saveAsTextFile(path)
将一个RDD的全部元素写入一个文本方式的本地文件,HDFS或其它任何Hadoop支持的存储系统中.(每行等于每个元素调用toString()的结果)
saveAsSequenceFile(path)
将一个RDD的全部元素写入一个二进制方式的本地文件,HDFS或其它任何Hadoop支持的存储系统中.
在Scala中，它还可以用于隐式转换为可写类型的类型(Spark包含对基本类型的转换，如Int、Double、String等)。
saveAsObjectFile(path)
将一个RDD的全部元素使用Java序列化以简单的格式编写数据集的元素(可以使用SparkContext.objectFile()加载这些元素)。
foreach(func)
在数据集的每个元素上运行函数func。这通常用于处理副作用，如更新累加器或与外部存储系统交互
注意:不可以修改foreach()之外的累加器之外的变量,见前面集群下的变量与闭包一节
1.4 Shuffle过程
Spark的某些操作,会引起一个Shuffle过程.Shuffle是指不同节点上的不同分区数据整合重新分区分组的机制.
所以Shuffle是一个代价很高的操作,因为它会导致executor和不同的机器节点之间进行数据复制.
1.4.1 Shuffle简述
以reduceByKey为例,将原始数据中key相同的记录聚合为一个组.这里挑战是原始数据很可能是存在不同分区不同机器的(参考MapReduce执行过程)
Spark-Shuffle与MapReduce-Shuffle的区别
MapReduce-Shuffle结果是分区有序,分区内再按Key排序
Spark-Shuffle结果是分区有序,但分区内Key无序.
要对Spark-Shuffle的分区内再排序,有以下方法:
mapPartitions 在已有的每个分区上再使用.sort排序
repartitionAndSortWithinPartitions 重建分区,并排序
sortBy提前对RDD本身做一个全范围排序
1.4.2 RDD中引起Shuffle的操作
repartition类操作 例如:repartition、coalesce
_ByKey操作（除了counting相关操作）例如：groupByKey、reduceByKey
join 例如：cogroup、join
1.4.3 Shuffle的性能影响
Shuffle本身是同时高耗内存,高耗磁盘IO,高耗网络IO的昂贵操作.
Spark会启动一系列的MapReduce(Hadoop MapReduce),产生大量的数据缓冲区与归并排序,大量的pill文件与归并Merge等等