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 :21scala> 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}}.collectres16: 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 ::= elempart_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 :21scala> 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

本文转自http://www.bubuko.com/infodetail-955333.html，所有权力归原作者所有。

Spark算子：统计RDD分区中的元素及数量相关推荐

如何统计二维或者多维空间/二维数组中重复元素的数量/计数？numpy一行代码就行了
文章目录 1. 问题说明 2. 解决之道 3. 向原作者致敬 Reference 1. 问题说明统计一维数组中重复元素的数量是个常见的问题,那么拓展到二维或者更高维度,这个问题依然是常见的.有什么方 ...
python统计各种数量_python 统计列表中不同元素的数量方法
python 统计列表中不同元素的数量方法刚刚上网搜了一下如何用python统计列表中不同元素的数量,发现很少,找了半天.我自己来写一种方法. 代码如下 list=[1,1,2,2,3] print ...
python输入一个字符串、计算其中小写字符的个数_利用键盘录入,输入一个字符串,统计该字符串中各个字符的数量,并输出（c/c++实现）...
title: 输入一串字符串统计字符个数 localimage: image1 urlname: staticstringnum categories: summarytags: [writen, e ...
求列表中个元素的数量
# 求列表中个元素的数量 lists = [1, 2, 3, 4, 5, 6, 7, 5, 4, 3, 2, 12] # 方法一:字典推导式(速度较快,优先推荐) dicts = {x: lists. ...
Arduino 统计中文字符串中的中文标点符号数量
Arduino 统计中文字符串中的中文标点符号数量先科普数据类型和所占内存空间大小一个英文字符等于1个字节,一个中文(含繁体)字符等于3个字节:中文标点占3个字节,英文标点占1个字节. 1Byte ...
Spark算子与RDD基本转换
map 将一个RDD中的每个数据项,通过map中的函数映射变为一个新的元素. 输入分区与输出分区一对一,即:有多少个输入分区,就有多少个输出分区. flatMap 属于Transformation算子 ...
Spark算子：RDD基本转换操作–coalesce、repartition
1. coalesce def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = n ...
Spark算子：RDD创建操作
关键字:Spark RDD 创建.parallelize.makeRDD.textFile.hadoopFile.hadoopRDD.newAPIHadoopFile.newAPIHadoopRDD ...
RDKit | 统计分子库中某种元素出现的次数
导入库 import pandas as pd from rdkit import Chem from rdkit.Chem import rdMolDescriptors from rdkit.Ch ...

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

Spark算子：统计RDD分区中的元素及数量相关推荐

最新文章

热门文章