如何使用Spark查找中位数和分位数

Question

如何RDD使用分布式方法，IPython和Spark查找整数的整数中位数？该RDD元素约为700,000个元素，因此太大而无法收集和找到中位数。

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

我知道我首先要排序RDD。我不知道怎么。我看到了sortBy（按给定的RDD排序keyfunc）和sortByKey（按RDD假定的（key，value）对组成的this排序）。我认为两者都使用键值，而我RDD只有整数元素。

1. 首先，我在想做什么myrdd.sortBy(lambda x: x)？
2. 接下来，我将找到rdd（rdd.count()）的长度。
3. 最后，我想在rdd的中心找到一个或两个元素。我也需要这种方法的帮助。

编辑：

我有个主意 也许我可以索引我的索引RDD，然后键=索引和值=元素。然后我可以尝试按值排序？我不知道这是否可能，因为只有一种sortByKey方法。

Answer 1

正在进行的工作

SPARK-30569 -
添加DSL功能调用percentile_approx

Spark 2.0+：

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

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聚合（全局和摸索）中：

> 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

火花<2.0

蟒蛇

正如我在评论中提到的那样，很可能不值得大惊小怪。如果像您这样的情况数据相对较小，则只需在本地收集和计算中位数即可：

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的内存。

如果数据更大，则排序将是一个限制因素，因此与其获得确切的值，不如直接在本地进行采样，收集和计算。但是，如果您真的想使用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)

和一些测试：

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 s，没有任何网络通信。可能有改善的方法，但是为什么还要打扰呢？

与语言无关 （ 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可以传递一个附加参数，该参数确定要使用的记录数。