Spark算子:RDD行动Action操作(4)–countByKey、foreach


转载:http://lxw1234.com/archives/2015/07/399.htm


遇到一个在spark shell上执行foreach什么都不显示的问题。


答案在下方:

countByKey

def countByKey(): Map[K, Long]

countByKey用于统计RDD[K,V]中每个K的数量。

  
  
  1. scala> var rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("B",3)))
  2. rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[7] at makeRDD at :21
  3.  
  4. scala> rdd1.countByKey
  5. res5: scala.collection.Map[String,Long] = Map(A -> 2, B -> 3)
  6.  

foreach

def foreach(f: (T) ⇒ Unit): Unit

foreach用于遍历RDD,将函数f应用于每一个元素。

但要注意,如果对RDD执行foreach,只会在Executor端有效,而并不是Driver端。

比如:rdd.foreach(println),只会在Executor的stdout中打印出来,Driver端是看不到的。

我在Spark1.4中是这样,不知道是否真如此。

 

这时候,使用accumulator共享变量与foreach结合,倒是个不错的选择。

  
  
  1. scala> var cnt = sc.accumulator(0)
  2. cnt: org.apache.spark.Accumulator[Int] = 0
  3.  
  4. scala> var rdd1 = sc.makeRDD(1 to 10,2)
  5. rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at makeRDD at :21
  6.  
  7. scala> rdd1.foreach(x => cnt += x)
  8.  
  9. scala> cnt.value
  10. res51: Int = 55
  11.  
  12. scala> rdd1.collect.foreach(println)
  13. 1
  14. 2
  15. 3
  16. 4
  17. 5
  18. 6
  19. 7
  20. 8
  21. 9
  22. 10
  23.  

foreachPartition

def foreachPartition(f: (Iterator[T]) ⇒ Unit): Unit

foreachPartition和foreach类似,只不过是对每一个分区使用f。

  
  
  1. scala> var rdd1 = sc.makeRDD(1 to 10,2)
  2. rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at makeRDD at :21
  3.  
  4. scala> var allsize = sc.accumulator(0)
  5. size: org.apache.spark.Accumulator[Int] = 0
  6.  
  7. scala> var rdd1 = sc.makeRDD(1 to 10,2)
  8. rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[6] at makeRDD at :21
  9.  
  10. scala> rdd1.foreachPartition { x => {
  11. | allsize += x.size
  12. | }}
  13.  
  14. scala> println(allsize.value)
  15. 10
  16.  

sortBy

def sortBy[K](f: (T) ⇒ K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

sortBy根据给定的排序k函数将RDD中的元素进行排序。

  
  
  1. scala> var rdd1 = sc.makeRDD(Seq(3,6,7,1,2,0),2)
  2.  
  3. scala> rdd1.sortBy(x => x).collect
  4. res1: Array[Int] = Array(0, 1, 2, 3, 6, 7) //默认升序
  5.  
  6. scala> rdd1.sortBy(x => x,false).collect
  7. res2: Array[Int] = Array(7, 6, 3, 2, 1, 0) //降序
  8.  
  9. //RDD[K,V]类型
  10. scala>var rdd1 = sc.makeRDD(Array(("A",2),("A",1),("B",6),("B",3),("B",7)))
  11.  
  12. scala> rdd1.sortBy(x => x).collect
  13. res3: Array[(String, Int)] = Array((A,1), (A,2), (B,3), (B,6), (B,7))
  14.  
  15. //按照V进行降序排序
  16. scala> rdd1.sortBy(x => x._2,false).collect
  17. res4: Array[(String, Int)] = Array((B,7), (B,6), (B,3), (A,2), (A,1))
  18.  

 

更多关于Spark算子的介绍,可参考 Spark算子系列文章 :

http://lxw1234.com/archives/2015/07/363.htm


关注微信公众号

注意!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系我们删除。



 
粤ICP备14056181号  © 2014-2020 ITdaan.com