RDD行动操作算子 --- foreach（遍历）、collect

RDD行动操作算子 --- foreach（遍历）、collect

foreach操作是直接调迭代rdd中每一条数据

class foreachTest { val conf = new SparkConf().setMaster("local[6]").setAppName("sortBy") val sc = new SparkContext(conf) @Test def foreachTest: Unit ={ val rdd = sc.parallelize(Seq(1,2,3,4,5,6)) rdd.foreach( item => println(item) ) }}

collect 在驱动程序中，以数组的形式返回数据集的所有元素

@Test def collectTest: Unit ={ sc.parallelize(Seq(("a",1),("b",2),("c",3))) .mapValues(item => item*10) .collect() .foreach(println(_)) }(a,10)(b,20)(c,30)

1.collect的作用 Spark内有collect方法，是Action操作里边的一个算子，这个方法可以将RDD类型的数据转化为数组， 同时会从远程集群是拉取数据到driver端。 2.已知的弊端 首先，collect是Action里边的，根据RDD的惰性机制，真正的计算发生在RDD的Action操作。 那么，一次collect就会导致一次Shuffle，而一次Shuffle调度一次stage，然而一次stage 包含很多个已分解的任务碎片Task。这么一来，会导致程序运行时间大大增加，属于比较 耗时的操作，即使是在local模式下也同样耗时。 其次，从环境上来讲，本机local模式下运行并无太大区别，可若放在分布式环境下运行， 一次collect操作会将分布式各个节点上的数据汇聚到一个driver节点上，而这么一来， 后续所执行的运算和操作就会脱离这个分布式环境而相当于单机环境下运行，这也与 Spark的分布式理念不合。 最后，将大量数据汇集到一个driver节点上，并且像这样val arr = data.collect()， 将数据用数组存放，占用了jvm堆内存，可想而知，是有多么轻松就会内存溢出。3.如何规避 若需要遍历RDD中元素，大可不必使用collect，可以使用foreach语句； 若需要打印RDD中元素，可用take语句，返回数据集前n个元素，data.take(1000).foreach(println)，这点官方文档里有说明； 若需要查看其中内容，可用saveAsTextFile方法。 总之，单机环境下使用collect问题并不大，但分布式环境下尽量规避，如有其他需要，手动编写代码实现相应功能就好。4.补充： collectPartitions：同样属于Action的一种操作，同样也会将数据汇集到Driver节点上，与 collect区别并不是很大，唯一的区别是：collectPartitions产生数据类型不同于collect，collect 是将所有RDD汇集到一个数组里，而collectPartitions是将各个分区内所有元素存储到一个数 组里，再将这些数组汇集到driver端产生一个数组；collect产生一维数组，而collectPartitions 产生二维数组。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。