如何使用分布式方法,IPython和Spark查找整数RDD
的中位数? RDD
大约为700,000个元素,因此太大而无法收集和找到中位数。
此问题类似于此问题。但是,问题的答案是使用Scala,我不知道。
How can I calculate exact median with Apache Spark?
使用Scala答案的思考,我试图用Python编写类似的答案。
我知道我首先想对RDD
进行排序。我不知道怎么。我看到了sortBy
(按给定的keyfunc
对RDD进行排序)和sortByKey
(对RDD
进行排序,并假定它们由(键,值)对组成。)方法。我认为两者都使用键值,而我的RDD
仅包含整数元素。
myrdd.sortBy(lambda x: x)
? rdd.count()
)的长度。编辑:
我有个主意。也许我可以索引我的RDD
,然后键=索引和值=元素。然后我可以尝试按值排序?我不知道是否可能,因为只有sortByKey
方法。
SPARK-30569-添加调用percentile_approx的DSL功能]
您可以使用实现approxQuantile
的Greenwald-Khanna algorithm方法:
Python
::df.approxQuantile("x", [0.5], 0.25)
Scala
df.stat.approxQuantile("x", Array(0.5), 0.25)
最后一个参数是相对误差。数字越低,结果越准确,计算量也就越多。
自Spark 2.2(SPARK-14352)起,它支持对多列进行估算:
df.approxQuantile(["x", "y", "z"], [0.5], 0.25)
和
df.approxQuantile(Array("x", "y", "z"), Array(0.5), 0.25)
[使用
approx_percentile
函数的SQL聚合(全局和摸索)中也可以使用基本方法:
approx_percentile
火花<2.0
Python
正如我在评论中提到的,很可能不值得大惊小怪。如果数据相对较小(如您的情况),则只需在本地收集和计算中位数:
> SELECT approx_percentile(10.0, array(0.5, 0.4, 0.1), 100); [10.0,10.0,10.0] > SELECT approx_percentile(10.0, 0.5, 100); 10.0
在我使用了几年的旧计算机上大约需要0.01秒,大约需要5.5MB的内存。
如果数据大得多,排序将是一个限制因素,因此与其获取确切值,不如直接在本地进行采样,收集和计算。但是,如果您真的想使用Spark,则应该使用类似的方法(如果我没有弄乱任何东西):
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)
到目前为止还不错,但是在本地模式下需要4.66 s,没有任何网络通信。可能有改善的方法,但是为什么还要打扰呢?
与语言无关
(Hive UDAF):如果使用from functools import partial
median = partial(quantile, p=0.5)
,您也可以使用Hive UDAF。带有整数值:
HiveContext
具有连续值:
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")
中,您可以传递一个附加参数,该参数确定要使用的记录数。
如果仅需要RDD方法而又不想移至DF,则添加解决方案。此摘录可以使您获得RDD两倍的百分位数。
这是我使用窗口函数(与pyspark 2.2.0一起使用的方法。)>
/**
* 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), e.g 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)
}
我已经编写了将数据帧作为输入并返回一个数据帧的函数,该数据帧的中位数作为分区上的输出,而order_col是我们要为其计算中值的列part_col是我们要计算的级别中位数: