首页 > 代码库 > 通过一系列破坏行为加深对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 版)