首页 > 代码库 > 通过一系列破坏行为加深对spark RDD 的理解(或者是猜测)(python 版)

通过一系列破坏行为加深对spark RDD 的理解(或者是猜测)(python 版)

这个实验由一个实验案例产生,实验中,需要对一个数据集进行维护,其中有一个需要对数据一条条进行插入:

下面是最二的写法:

rdd=sc.parallelize([-1])for i in range(10000):    rdd=rdd.union(sc.parallelize([i]))

每次插入数据,新建一个rdd,然后union。

后果是:

java.lang.OutOfMemoryError: GC overhead limit exceeded

at org.apache.spark.rdd.UnionRDD$$anonfun$getPartitions$2$$anonfun$apply$1.apply(UnionRDD.scala:69)
at org.apache.spark.rdd.UnionRDD$$anonfun$getPartitions$2$$anonfun$apply$1.apply(UnionRDD.scala:68)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
at org.apache.spark.rdd.UnionRDD$$anonfun$getPartitions$2.apply(UnionRDD.scala:68)
at org.apache.spark.rdd.UnionRDD$$anonfun$getPartitions$2.apply(UnionRDD.scala:68)

。。。

第2119次循环时报的错,但不能对rdd进行操作。也就是说看不出来成功插入的条数,反正就是没意义。

方法二:也挺二,其实是第一种二的基础上改进一下

count=0rdd=sc.parallelize([-1])for i in range(10000):    rdd=rdd.union(sc.parallelize([i]))    count=count+1    if(count>100):        rdd.take(1)        count=0

: org.apache.spark.SparkException: Job aborted due to stage failure: Task serialization failed: java.lang.StackOverflowError
java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)

第605次循环就报错了,一样的二,类似的结果,不一样的原因

也许还有第三个实验:

count=0rdd=sc.parallelize([-1])for i in range(10000):    rdd=rdd.union(sc.parallelize([i])).persist()    count=count+1    if(count>100):        rdd.take(1)        count=0

跟实验二差不多。

现在说一下原因。

首先类比一下我们在内存中干同样的事情,

我们有一个数组,循环一万次,每次插入一个元素,似乎没有任何问题。我们也经常这么做。

那我们经常把rdd看作数组(或者list)对待,现在要插入一个元素,似乎只能是用union。为什么就不行?

RDD将操作分为两类:transformation与action。无论执行了多少次transformation操作,RDD都不会真正执行运算,只有当action操作被执行时,运算才会触发。而在RDD的内部实现机制中,底层接口则是基于迭代器的,从而使得数据访问变得更高效,也避免了大量中间结果对内存的消耗。

可见rdd于数组还是有区别。在问题一中,union都被记录下来,而没有去执行,然后,就把内存撑爆了。

第二次,我每过100次,就执行一次action,这样,union就会被执行了,好聪明啊。

但还是挂了,而且挂的更惨。

这是因为,计算的时候,rdd并不把上次rdd的结果作为输入,而是,依然把最初的数据装进来,从头开始算,但是,要算的东西太多,所以堆栈溢出。

第三次,表明persist操作并无明显的作用。

好吧,突然又想到实验四。

count=0rdd=sc.parallelize([-1])for i in range(10000):    rdd=rdd.union(sc.parallelize([i])).persist()    count=count+1    if(count>100):        myarray=rdd.collect()        rdd=sc.parallelize(myarray)        count=0

结果还在跑,但感觉问题不大。

就是这样的话,数据量过大的话,myarray还是把内存撑爆啊。

 

综上:最好的做法是,在内存里攒着别插,然后攒够一万条就插一次。

而且在别的场景中,你可以制造大规模的数据,但不要制造大规模的transformation操作。

 

通过一系列破坏行为加深对spark RDD 的理解(或者是猜测)(python 版)