spark 查找 中位数

基本的想法是,

第一步先partition,把一定范围的数字放入同一个partition

第二步,记录每一个partition的size,可以知道中位数落在哪个partition里面

第三步,把第二部找到的partition进行toArray,然后sort,然后获取中位数,或者不sort,利用的取中位数的函数直接取(后面博客会介绍)

先准备数据:

import java.io.PrintWriter

val max=9999999
val parts=10

val out=new PrintWriter("number")
for (i <- 1 to max){
out.println(i)
}
out.close

加载数据,进行上面的三不处理:
//加载数据
val file = sc.textFile("file:/root/number")
//创建分区函数
 import org.apache.spark.Partitioner
val parter = new Partitioner {
def getPartition(key: Any): Int = key match {
case i: Int => i / (max / parts + 1)
case _ => parts - 1
}

val numPartitions = parts
}
//分区,应该可以用partition函数,有空试试
val g = file.groupBy((x: String) => x.toInt, parter).cache
//计算每个partition的大小
 val arr_partition = g.mapPartitions(x => Iterator(x.size)).collect
//计算中位数落在哪个分区,并计算是第几个值
val harf = (file.count) / 2
var count = 0
var index = 0
var flag = true
for (i <- arr_partition if flag) {
count += i
println(i)
if (count > harf) {
flag = false
count -= i
} else {
index += 1
}
}

//计算出中位数,array中不为了0的那个就是,或者直接获取array的第index位置的值,就是中位数
g.mapPartitionsWithIndex((i, x) => {
if (i != index) {
Iterator(0)
} else {
val tmp = x.toArray.sorted
val (k, _) = tmp((harf - count).toInt)
Iterator(k)
}
}
).collect不用并行处理,用快排的思想,应该也能很快取到中位数,下次再写