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

我怎样才能找到一个中位数RDD使用分布式方法、IPython 和 Spark 计算整数？这RDD大约有 700,000 个元素，因此太大而无法收集和查找中位数。

这个问题与这个问题类似：如何使用 Apache Spark 计算精确中位数？ https://stackoverflow.com/questions/28158729/how-can-i-calculate-exact-median-with-apache-spark。然而，问题的答案是在 Scala 中，我不知道。

使用 Scala 答案的想法，我尝试用 Python 编写一个类似的解决方案。

我知道我首先要排序RDD。我不知道怎么。我看到sortBy（按给定的值对该 RDD 进行排序keyfunc) and sortByKey（对此进行排序RDD，假设由（键，值）对组成。）方法。我认为两者都使用键值和我的RDD仅具有整数元素。

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

EDIT:

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

正在进行的工作

SPARK-30569 https://issues.apache.org/jira/browse/SPARK-30569 - 添加调用percentile_approx 的DSL 函数

火花2.0+：

您可以使用approxQuantile实现的方法格林沃尔德-卡纳算法 http://infolab.stanford.edu/~datar/courses/cs361a/papers/quantiles.pdf:

Python:

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

Scala:

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

其中最后一个参数是相对误差。数字越低，结果越准确，计算成本也越高。

从 Spark 2.2 开始（SPARK-14352 https://issues.apache.org/jira/browse/SPARK-14352）它支持对多列进行估计：

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

and

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

底层方法也可用于 SQL 聚合（全局和摸索）：approx_percentile https://spark.apache.org/docs/latest/api/sql/index.html#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

火花

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 秒。可能有办法改善这一点，但为什么还要麻烦呢？

独立于语言 (蜂巢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")

In percentile_approx您可以传递一个附加参数来确定要使用的记录数。