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

我怎样才能find使用分布式方法，IPython和Spark整数RDD的中位数？ RDD大约有70万个元素，因此太大而无法收集和find中位数。

这个问题类似于这个问题。 但是，问题的答案是使用我不知道的Scala。

我如何用Apache Spark计算确切的中位数？

使用Scala的思考答案，我试图用Python编写一个类似的答案。

我知道我首先要对RDD进行sorting。 我不知道怎么。 我看到sortBy （通过给定的keyfunc对此RDD进行sorting）和sortByKey （对此RDDsorting，假定它由（键，值）对组成）方法。 我认为这两个使用键值，我的RDD只有整数元素。

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

编辑：

我有一个想法。 也许我可以索引我的RDD ，然后key = index和value = element。 然后我可以尝试按价值sorting？ 我不知道这是否可能，因为只有一个sortByKey方法。

Spark 2.0+：

您可以使用实现Greenwald-Khannaalgorithm的 approxQuantile方法：

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

我几年的电脑和大约5.5MB的内存大约需要0.01秒。

如果数据大得多，sorting将成为一个限制因素，所以不是获取确切的值，而是在本地进行采样，收集和计算。 但如果你真的想要使用Spark这样的应该做的伎俩（如果我没有搞砸任何东西）： 

 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)

还有一些testing： 

 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)

到目前为止这么好，但在没有任何networking通信的情况下，在本地模式下需要4.66秒。 可能有办法来改善这一点，但为什么还要麻烦？

独立语言 （ Hive UDAF ）：

如果你使用HiveContext你也可以使用Hive UDAF。 具有整数值： 

 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数量。

如果您只需要一个RDD方法并且不想移动到DF，则添加一个解决scheme。 这个片段可以让你得到一个RDD的百分位数。

如果你input百分位数为50，你应该获得你所要求的中位数。 让我知道是否有任何angular落案件没有说明。 

 /** * 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) }

这里是我使用窗口函数的方法（使用pyspark 2.2.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

然后调用addMedian方法来计算col2的中值： 

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

最后你可以按需要分组。 

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