如何使用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方法。

3 Solutions collect form web for “如何使用Sparkfind中位数和分位数”

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") 
  • caching和持久性有什么区别?
  • 如何使用spark从hbase读取
  • Spark - 重新分区()vs coalesce()