Spark使用总结

1）spark通常把shuffle操作定义为划分stage的边界，其实stage的边界有两种：ShuffleMapTask和ResultTask。ResultTask就是输出结果，输出结果的称为ResultTask，都为引起stage的划分，比如以下代码：

rdd.parallize(1 to 10).foreach(println)

每个stage内部，一定有一个ShuffleMapTask或者是ResultTask，因为这两者是划分stage的依据，是stage之间的边界。一个stage中的所有task最后会以taskSet的形式提交给TaskScheduler去执行，Spark实现了三种不同的

TaskScheduler

，包括

LocalSheduler

、

ClusterScheduler

和

MesosScheduler

。

2）actions(动作)会生成一个job，触发job的提交，所以我们从客户端提交的一个作业可能会被划分为多个job。但是，如果一个action后没有其他操作，也就是这个action是最后一个操作的话，这个action就独立为一个stage，而非提交一个job。（参考0）

3)task分为ShuffleMapTask和ResultTask（参考1）。

4）广播变量应该广播RDD对应的值，也就是广播RDD.collect()而不是RDD本身，可以使用spark-shell测试。（参考2）

5）使用mapPartitions以分区为单位处理RDD时，定义的函数返回值必须是Iterator类型，同样的，mapPartitionsWithIndex，也必须返回Iterator类型。mapPartitions（iter:Iterator[]）有一个Iterator类型的参数，使用以下方式使用mapPartitions：

val rdd1=....
val rdd2=rdd1.mapPartitions(processPartitions)//如果我们定义的函数除了Iterator之外，没有别的参数，就可以以这种方式来使用mapPartitions

val rdd3=rdd1.mapPartitions((x,y)=>processPartitions2(x,y))//其中x是Iterator类型，y是broad

//定义processPartitions
.mapPartitions只有一个Iterator参数，所以我们自定义的函数也必须包含这个参数
def processPartitions(iter:Iterator[Int]){//假设分区中每个元素类型是Int
...

}

def processPartitions2(iter:Iterator[Int],broad:Array[Double]){//假设分区中每个元素类型是Int
...

}

mapPartitionsIndex的使用与mapPartitions类似，只是mapPartitionsIndex(Idx:Int,iter:Iterator[U])多了一个参数Idx，Idx就是分区的编号。参考3

6)使用rdd1.zipPartitions(rdd2)(substractTwoPatitionsValue),在iter.hasNext之前，不能使用调用iter变量做任何的操作，这会导致iter变为空，等到使用iter.hasNext时取不到值，错误代码如下：

def substractTwoPatitionsValue(iter1:Iterator[(Int,Array[Double])],iter2:Iterator[(Int,Array[Double])]): Iterator[(Int,Array[Double])] ={
val matRows=iter1.length//这里在hasNext之前使用Iter1，可以得到正确值
println("matRows:"+matRows)//522
println("iter1 length:"+iter1.length)//第二次使用iter就不能取到值了，结果是0
//println("iter2 length:"+iter2.length)
val resMatrixColumns=21025//21025
val u=Array.ofDim[Double](matRows,resMatrixColumns)
val keyIndexArray=new Array[Int](matRows)//分区中的key值
var keyCounter=(-1)
//var keyIndex=0
var Value=new Array[Double](resMatrixColumns)
var cur:(Int,Array[Double])=(0,new Array[Double](resMatrixColumns))
println("before iter1")
while(iter1.hasNext){
keyCounter+=1
println("in iter1")
cur=iter1.next()
keyIndexArray(keyCounter)=cur._1//这里每个分区里的key/value是按照key逆序排列
Value=cur._2
for(k<-0 until resMatrixColumns){//right matrix cols
u(keyCounter)(k)=Value(k)//将顺序读取到的分区里的值，存储到u中
}
//println(cur._2(1))
}
/************************以上就完成了读取第1个rdd中1个分区的数据的任务*************************************************************/

/************************接下来读取第2个rdd中1个分区的数据,并根据key和第1个rdd中分区值相加************************************************************/
//var cc:(Int,Array[Double])=(0,new Array[Double](resMatrixColumns))
var keyIndex=0
var positionInMatrixu=0
println("before iter2")
while(iter2.hasNext){
println("in iter2")
cur=iter2.next()
keyIndex=cur._1
Value=cur._2
positionInMatrixu=keyIndexArray.indexOf(keyIndex)//存储数据时，keyIndexArray和u的索引值通过keyCounter一样对应
for(k<-0 until resMatrixColumns){//right matrix cols
u(positionInMatrixu)(k)-=Value(k)//完成两个分区值的相加
}
}
val newIter=keyIndexArray.zip(u).iterator //结果是有序的，不像SRdd中key是无序的
newIter
/* 使用以上方式产生迭代子
keyCounter=(-1)
val newIter=u.map(x=>{
keyCounter+=1
//println("keyIndex:"+keyIndex)
(keyIndexArray(keyCounter),x)}).toIterator
newIter
*/
}

也就是说，在zipPartitions里，各个RDD的迭代子iter1以下面这种形式使用，并且只能使用一次：

while(iter1.hasNext){
.....
｝

7）运行spark作业的时候有时候会出现“Task deserialization Time”过长的问题，如下所示：



也就是“任务反序列化时间”过长，原因是调用rdd的转换操作时，使用了“外部变量”（driver中的一个数组），解决方法是使用“广播变量”将此“外部变量”进行广播。

8）同步、异步、阻塞、非阻赛：参考