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

Question

如何使用分布式方法、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 只有整数元素。

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

编辑:

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

Answer 1

正在进行的工作

SPARK-30569 - 添加调用 percentile_approx 的 DSL 函数

Spark 2.0+:

您可以使用实现 Greenwald-Khanna algorithm 的 approxQuantile 方法:

Python:

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

斯卡拉:

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)

底层方法也可用于 SQL 聚合(全局和摸索)，使用 approx_percentile功能:

> 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

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 秒。

如果数据大得多，排序将是一个限制因素，因此与其获取精确值，不如在本地进行采样、收集和计算。但是如果你真的想使用 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 秒。可能有办法改善这一点，但为什么还要麻烦呢？

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