如何使用星火找到中位数和分位数

Question

如何使用星火找到中位数和分位数我怎样才能找到RDD使用分布式方法的整数，IPython和SPark？这个RDD大约有70万个元素，因此太大，无法收集和找到中位数。这个问题和这个问题类似。但是，这个问题的答案是使用Scala，我不知道。如何计算ApacheSPark的精确中值？使用Scala答案的思路，我试图用Python编写类似的答案。我知道我首先想把RDD..我不知道怎么做。我看到sortBy(将此rdd按给定的keyfunc)和sortByKey(整理这个RDD，它假定由(键，值)对组成。)方法。我认为两者都使用键值，而我的RDD只有整数元素。首先，我在考虑myrdd.sortBy(lambda x: x)?接下来，我将找到RDD的长度(rdd.count()).最后，我希望找到RDD中心的元素或2个元素。这个方法我也需要帮助。编辑：我有个主意。也许我可以索引我的RDD然后Key=index和value=Element。然后我就可以试着按价值分类了？我不知道这是否可能，因为只有一个sortByKey方法。

Answer 1

火花2.0+：

你可以用approxQuantile实现Greenwald-Khanna算法:

Python:

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

斯卡拉:

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

其中最后一个参数是一个相对错误。次数越少，计算结果越准确，计算量越大。

星火2.2(火花-14352)它支持对多列的估计：

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

和

df.approxQuantile(Array("x", "y", "z"), Array(0.5), 0.25)

火花<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

在我几年前的电脑上，它需要大约0.01秒的时间和大约5.5MB的内存。

如果数据要大得多，排序将是一个限制因素，因此，与其获得确切的值，不如在本地进行采样、收集和计算。但是，如果你真的想让一个人使用星火，这样的事情应该能起作用(如果我什么都没搞砸的话)：

from numpy import floorimport timedef 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)

到目前为止还不错，但是它需要4.66秒的本地模式，没有任何网络通信。也许有办法改善这一点，但为什么还要费心呢？

语言独立 (蜂箱):

如果你用HiveContext您也可以使用HiveUDAFs。具有积分值：

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您可以传递另一个参数，该参数确定要使用的多个记录。

Answer 2

如果您只想要一个RDD方法，并且不想移动到DF，那么添加一个解决方案。这个片段可以为双的RDD获得一个百分位数。

如果输入百分位数为50，则应获得所需的中位数。如果有不明案件，请告诉我。

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