# 如何使用Sparkfind中位数和分位数

1. 首先，我正在考虑做`myrdd.sortBy(lambda x: x)`
2. 接下来我会findrdd（ `rdd.count()` ）的长度。
3. 最后，我想在rdd的中心find元素或2个元素。 我也需要这个方法的帮助。

### Spark 2.0+：

Python

` `df.approxQuantile("x", [0.5], 0.25)` `

` `df.stat.approxQuantile("x", Array(0.5), 0.25)` `

### Spark <2.0

python

` `import numpy as np np.random.seed(323) rdd = sc.parallelize(np.random.randint(1000000, size=700000)) %time np.median(rdd.collect()) np.array(rdd.collect()).nbytes` `

` `from numpy import floor import time def quantile(rdd, p, sample=None, seed=None): """Compute a quantile of order p ∈ [0, 1] :rdd a numeric rdd :p quantile(between 0 and 1) :sample fraction of and rdd to use. If not provided we use a whole dataset :seed random number generator seed to be used with sample """ assert 0 <= p <= 1 assert sample is None or 0 < sample <= 1 seed = seed if seed is not None else time.time() rdd = rdd if sample is None else rdd.sample(False, sample, seed) rddSortedWithIndex = (rdd. sortBy(lambda x: x). zipWithIndex(). map(lambda (x, i): (i, x)). cache()) n = rddSortedWithIndex.count() h = (n - 1) * p rddX, rddXPlusOne = ( rddSortedWithIndex.lookup(x)[0] for x in int(floor(h)) + np.array([0L, 1L])) return rddX + (h - floor(h)) * (rddXPlusOne - rddX)` `

` `np.median(rdd.collect()), quantile(rdd, 0.5) ## (500184.5, 500184.5) np.percentile(rdd.collect(), 25), quantile(rdd, 0.25) ## (250506.75, 250506.75) np.percentile(rdd.collect(), 75), quantile(rdd, 0.75) (750069.25, 750069.25)` `

` `from functools import partial median = partial(quantile, p=0.5)` `

` `rdd.map(lambda x: (float(x), )).toDF(["x"]).registerTempTable("df") sqlContext.sql("SELECT percentile_approx(x, 0.5) FROM df")` `

` `sqlContext.sql("SELECT percentile(x, 0.5) FROM df")` `

`percentile_approx`您可以传递一个额外的参数来确定要使用的logging数量。

` `/** * Gets the nth percentile entry for an RDD of doubles * * @param inputScore : Input scores consisting of a RDD of doubles * @param percentile : The percentile cutoff required (between 0 to 100), eg 90%ile of [1,4,5,9,19,23,44] = ~23. * It prefers the higher value when the desired quantile lies between two data points * @return : The number best representing the percentile in the Rdd of double */ def getRddPercentile(inputScore: RDD[Double], percentile: Double): Double = { val numEntries = inputScore.count().toDouble val retrievedEntry = (percentile * numEntries / 100.0 ).min(numEntries).max(0).toInt inputScore .sortBy { case (score) => score } .zipWithIndex() .filter { case (score, index) => index == retrievedEntry } .map { case (score, index) => score } .collect()(0) }` `

` `class median(): """ Create median class with over method to pass partition """ def __init__(self, df, col, name): assert col self.column=col self.df = df self.name = name def over(self, window): from pyspark.sql.functions import percent_rank, pow, first first_window = window.orderBy(self.column) # first, order by column we want to compute the median for df = self.df.withColumn("percent_rank", percent_rank().over(first_window)) # add percent_rank column, percent_rank = 0.5 coressponds to median second_window = window.orderBy(pow(df.percent_rank-0.5, 2)) # order by (percent_rank - 0.5)^2 ascending return df.withColumn(self.name, first(self.column).over(second_window)) # the first row of the window corresponds to median def addMedian(self, col, median_name): """ Method to be added to spark native DataFrame class """ return median(self, col, median_name) # Add method to DataFrame class DataFrame.addMedian = addMedian` `

` `median_window = Window.partitionBy("col1") df = df.addMedian("col2", "median").over(median_window)` `

` `df.groupby("col1", "median")` `
Interesting Posts