当前位置:首页 > 开发 > 开源软件 > 正文

Spark算子:统计RDD分区中的元素及数量

发表于: 2015-07-06   作者:superlxw1234   来源:转载   浏览:
摘要: 关键字:Spark算子、Spark RDD分区、Spark RDD分区元素数量     Spark RDD是被分区的,在生成RDD时候,一般可以指定分区的数量,如果不指定分区数量,当RDD从集合创建时候,则默认为该程序所分配到的资源的CPU核数,如果是从HDFS文件创建,默认为文件的Block数。   可以利用RDD的mapPartitionsWithInd

关键字:Spark算子、Spark RDD分区、Spark RDD分区元素数量

 

 

Spark RDD是被分区的,在生成RDD时候,一般可以指定分区的数量,如果不指定分区数量,当RDD从集合创建时候,则默认为该程序所分配到的资源的CPU核数,如果是从HDFS文件创建,默认为文件的Block数。

 

可以利用RDD的mapPartitionsWithIndex方法来统计每个分区中的元素及数量。

 

关于mapPartitionsWithIndex的介绍可以参考 mapPartitionsWithIndex的介绍

 

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

 

 

具体看例子:

 

 

//创建一个RDD,默认分区15个,因为我的spark-shell指定了一共使用15个CPU资源
//–total-executor-cores 15

 

 

scala> var rdd1 = sc.makeRDD(1 to 50)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[17] at makeRDD at :21
 
scala> rdd1.partitions.size
res15: Int = 15

 

 

//统计rdd1每个分区中元素数量

rdd1.mapPartitionsWithIndex{
        (partIdx,iter) => {
          var part_map = scala.collection.mutable.Map[String,Int]()
            while(iter.hasNext){
              var part_name = "part_" + partIdx;
              if(part_map.contains(part_name)) {
                var ele_cnt = part_map(part_name)
                part_map(part_name) = ele_cnt + 1
              } else {
                part_map(part_name) = 1
              }
              iter.next()
            }
            part_map.iterator
           
        }
      }.collect

res16: Array[(String, Int)] = Array((part_0,3), (part_1,3), (part_2,4), (part_3,3), 
(part_4,3), (part_5,4), (part_6,3), (part_7,3), (part_8,4), (part_9,3), (part_10,3), 
(part_11,4), (part_12,3), (part_13,3), (part_14,4))

//从part_0到part_14,每个分区中的元素数量

 

//统计rdd1每个分区中有哪些元素

 

rdd1.mapPartitionsWithIndex{
        (partIdx,iter) => {
          var part_map = scala.collection.mutable.Map[String,List[Int]]()
            while(iter.hasNext){
              var part_name = "part_" + partIdx;
              var elem = iter.next()
              if(part_map.contains(part_name)) {
                var elems = part_map(part_name)
                elems ::= elem
                part_map(part_name) = elems
              } else {
                part_map(part_name) = List[Int]{elem}
              }
            }
            part_map.iterator
           
        }
      }.collect
res17: Array[(String, List[Int])] = Array((part_0,List(3, 2, 1)), (part_1,List(6, 5, 4)), 
(part_2,List(10, 9, 8, 7)), (part_3,List(13, 12, 11)), (part_4,List(16, 15, 14)), 
(part_5,List(20, 19, 18, 17)), (part_6,List(23, 22, 21)), (part_7,List(26, 25, 24)), 
(part_8,List(30, 29, 28, 27)), (part_9,List(33, 32, 31)), (part_10,List(36, 35, 34)), 
(part_11,List(40, 39, 38, 37)), (part_12,List(43, 42, 41)), (part_13,List(46, 45, 44)), 
(part_14,List(50, 49, 48, 47)))
//从part_0到part14,每个分区中包含的元素

 

//从HDFS文件创建的RDD,包含65个分区,因为该文件由65个Block

scala> var rdd2 = sc.textFile("/logs/2015-07-05/lxw1234.com.log")
rdd2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[21] at textFile at :21
 
scala> rdd2.partitions.size
res18: Int = 65

 

//rdd2每个分区的元素数量

 

scala> rdd2.mapPartitionsWithIndex{
     |         (partIdx,iter) => {
     |           var part_map = scala.collection.mutable.Map[String,Int]()
     |             while(iter.hasNext){
     |               var part_name = "part_" + partIdx;
     |               if(part_map.contains(part_name)) {
     |                 var ele_cnt = part_map(part_name)
     |                 part_map(part_name) = ele_cnt + 1
     |               } else {
     |                 part_map(part_name) = 1
     |               }
     |               iter.next()
     |             }
     |             part_map.iterator
     |            
     |         }
     |       }.collect
res19: Array[(String, Int)] = Array((part_0,202496), (part_1,225503), (part_2,214375), 
(part_3,215909), (part_4,208941), (part_5,205379), (part_6,207894), (part_7,209496), 
(part_8,213806), (part_9,216962), (part_10,216091), (part_11,215820), (part_12,217043), 
(part_13,216556), (part_14,218702), (part_15,218625), (part_16,218519), (part_17,221056), 
(part_18,221250), (part_19,222092), (part_20,222339), (part_21,222779), (part_22,223578), 
(part_23,222869), (part_24,221543), (part_25,219671), (part_26,222871), (part_27,223200), 
(part_28,223282), (part_29,228212), (part_30,223978), (part_31,223024), (part_32,222889), 
(part_33,222106), (part_34,221563), (part_35,219208), (part_36,216928), (part_37,216733), 
(part_38,217214), (part_39,219978), (part_40,218155), (part_41,219880), (part_42,215833...

 

更多关于Spark算子的介绍,可参考 Spark算子

http://lxw1234.com/archives/tag/spark%E7%AE%97%E5%AD%90

 

Spark算子:统计RDD分区中的元素及数量

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
Key-Value型Transformation算子 Transformation处理的数据为Key-Value形式的算子,大致可以分为3种
Value型Transformation算子 处理数据类型为Value型的Transformation算子可以根据RDD变换算子的输入
Actions算子 本质上在Actions算子中通过SparkContext执行提交作业的runJob操作,触发了RDD DAG的执
本质上在Actions算子中通过SparkContext执行提交作业的runJob操作,触发了RDD DAG的执行。 根据Acti
算子运行过程 下图描述了Spark在运行转换中通过算子对RDD进行转换。 算子是RDD中定义的函数,可以对
1.interSection 2.join 1.interSection 1.示例代码 package spark.examples import org.apache.spar
1.distinct 2.cogroup 1.distinct 1.示例代码 package spark.examples import org.apache.spark.{Sp
1.count 2.groupByKey 3.join 4.union 5.reduceByKey Shuffle/Dependency总结 ShuffleMapTask将数据
1.count 2.groupByKey 3.join 4.union 5.reduceByKey Shuffle/Dependency总结 ShuffleMapTask将数据
1.count 2.groupByKey 3.join 4.union 5.reduceByKey Shuffle/Dependency总结 ShuffleMapTask将数据
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号