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

Question

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

85

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

这个问题类似于这个问题：如何使用Apache Spark计算精确的中位数？。然而，问题的答案是用Scala写的，而我不懂Scala。

根据Scala答案的思路，我正在尝试用Python编写类似的解决方案。

我知道我首先想要对RDD进行排序，但我不知道如何做。我看到了sortBy（按给定的keyfunc对RDD进行排序）和sortByKey（对RDD进行排序，假设它由(key, value)对组成）方法。我认为这两种方法都使用键值，而我的RDD只有整数元素。

首先，我在考虑执行 `myrdd.sortBy(lambda x: x)`？

接下来，我将找到 RDD 的长度（`rdd.count()`）。

最后，我想要找到 RDD 中位于中心位置的一个或两个元素。对于这个方法，我需要帮助。

编辑：

我有一个想法。也许我可以给我的 RDD 加上索引，然后将索引作为键，元素作为值。然后我可以尝试按值进行排序？我不知道是否可能，因为只有一个 `sortByKey` 方法。

- pr338

4

假设有700,000个整数，每个整数占用64位，需要约5MB的存储空间，使用np.median可以在本地计算中位数并在几分之一秒内完成。当然，您可以按照您所描述的进行排序和索引，但我猜它会慢大约一个数量级。 - zero323

6

也许这是在一组Commodore 64上运行的Spark集群。 - Daniel Darabos

2

@DanielDarabos 那是一个非常棒的想法 :) 并将磁带机用作HDFS替代方案... - zero323

以下是如何使用Pyspark Dataframe AP进行操作的说明：https://dev59.com/TZnga4cB1Zd3GeqPZoPD#38743477 - gench

RDD 代表什么？ - GenDemo

@GenDemo RDD 在 Spark 中代表弹性分布式数据集。 - Fony Lew

9个回答

12

这是我使用窗口函数的方法（使用pyspark 2.2.0）。

from pyspark.sql import DataFrame

class median():
    """ Create median class with over method to pass partition """
    def __init__(self, df, col, name):
        assert col
        self.column=col
        self.df = df
        self.name = name

    def over(self, window):
        from pyspark.sql.functions import percent_rank, pow, first

        first_window = window.orderBy(self.column)                                  # first, order by column we want to compute the median for
        df = self.df.withColumn("percent_rank", percent_rank().over(first_window))  # add percent_rank column, percent_rank = 0.5 coressponds to median
        second_window = window.orderBy(pow(df.percent_rank-0.5, 2))                 # order by (percent_rank - 0.5)^2 ascending
        return df.withColumn(self.name, first(self.column).over(second_window))     # the first row of the window corresponds to median

def addMedian(self, col, median_name):
    """ Method to be added to spark native DataFrame class """
    return median(self, col, median_name)

# Add method to DataFrame class
DataFrame.addMedian = addMedian

然后调用 addMedian 方法计算 col2 的中位数：

from pyspark.sql import Window

median_window = Window.partitionBy("col1")
df = df.addMedian("col2", "median").over(median_window)

最后，如果需要，您可以进行分组。

df.groupby("col1", "median")

- Benoît Carne

3

对于偶数个数字，这种方法行不通：中位数会出现问题。必须取中间两个元素的平均值作为中位数。 - Ihor B.

@BenoîtCarne 这个 DataFrame.addMedian = addMedian 的代码行是如何工作的? 在Python中它叫什么? - Shankar

我喜欢你如何将方法添加到DataFrame中。但是，Ihor的评论仍然是有效的 - 中位数是近似值而不是正确的值。 - ZygD

@ZygD 是的，你说得对。目前我还没有解决方法。不过对于我的使用情况来说，这个近似已经足够了。 - Benoît Carne

1

@Shankar 通常被称为猴子补丁："在Ruby、Python和许多其他动态编程语言中，猴子补丁这个术语仅指运行时对类或模块进行动态修改，旨在修补现有的第三方代码，以解决不按预期工作的错误或功能问题。" - eggsyntax

显示剩余3条评论

8

如果您希望仅使用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)
  }

- Vedant

3

有两种方法可以使用。一种是使用approxQuantile方法，另一种是使用percentile_approx方法。然而，当记录数为偶数时，这两种方法都可能无法给出准确的结果。解决方法是按下面所示取平均值。

importpyspark.sql.functions.percentile_approx as F
# df.select(F.percentile_approx("COLUMN_NAME_FOR_WHICH_MEDIAN_TO_BE_COMPUTED", 0.5).alias("MEDIAN)) # might not give proper results when there are even number of records

((
df.select(F.percentile_approx("COLUMN_NAME_FOR_WHICH_MEDIAN_TO_BE_COMPUTED", 0.5) + df.select(F.percentile_approx("COLUMN_NAME_FOR_WHICH_MEDIAN_TO_BE_COMPUTED", 0.500001)
)*.5).alias("MEDIAN))

- prashanth

2

我已经编写了一个函数，它以数据框作为输入，并返回一个数据框，其中包含对分区的中位数作为输出，order_col是我们希望计算part_col中位数的列，part_col是我们希望计算中位数的级别：

from pyspark.sql import Window
import pyspark.sql.functions as F

def calculate_median(dataframe, part_col, order_col):
    win = Window.partitionBy(*part_col).orderBy(order_col)
#     count_row = dataframe.groupby(*part_col).distinct().count()
    dataframe.persist()
    dataframe.count()
    temp = dataframe.withColumn("rank", F.row_number().over(win))
    temp = temp.withColumn(
        "count_row_part",
        F.count(order_col).over(Window.partitionBy(part_col))
    )
    temp = temp.withColumn(
        "even_flag",
        F.when(
            F.col("count_row_part") %2 == 0,
            F.lit(1)
        ).otherwise(
            F.lit(0)
        )
    ).withColumn(
        "mid_value",
        F.floor(F.col("count_row_part")/2)
    )

    temp = temp.withColumn(
        "avg_flag",
        F.when(
            (F.col("even_flag")==1) &
            (F.col("rank") == F.col("mid_value"))|
            ((F.col("rank")-1) == F.col("mid_value")),
            F.lit(1)
        ).otherwise(
        F.when(
            F.col("rank") == F.col("mid_value")+1,
            F.lit(1)
            )
        )
    )
    temp.show(10)
    return temp.filter(
        F.col("avg_flag") == 1
    ).groupby(
        part_col + ["avg_flag"]
    ).agg(
        F.avg(F.col(order_col)).alias("median")
    ).drop("avg_flag")

- Ankit Kumar Namdeo

0

我们可以使用以下代码在Spark中计算中位数和分位数：

df.stat.approxQuantile(col,[quantiles],error)

例如，在以下数据框中查找中位数[1,2,3,4,5]：

df.stat.approxQuantile(col,[0.5],0)

误差越小，结果越准确。

- Arun Nandam

0

要进行精确中位数计算，您可以使用以下函数，并将其与 PySpark DataFrame API 一起使用：

def median_exact(col: Union[Column, str]) -> Column:
    """
    For grouped aggregations, Spark provides a way via pyspark.sql.functions.percentile_approx("col", .5) function,
    since for large datasets, computing the median is computationally expensive.
    This function manually computes the median and should only be used for small to mid sized datasets / groupings.
    :param col: Column to compute the median for.
    :return: A pyspark `Column` containing the median calculation expression
    """
    list_expr = F.filter(F.collect_list(col), lambda x: x.isNotNull())
    sorted_list_expr = F.sort_array(list_expr)
    size_expr = F.size(sorted_list_expr)

    even_num_elements = (size_expr % 2) == 0
    odd_num_elements = ~even_num_elements

    return F.when(size_expr == 0, None).otherwise(
        F.when(odd_num_elements, sorted_list_expr[F.floor(size_expr / 2)]).otherwise(
            (
                sorted_list_expr[(size_expr / 2 - 1).cast("long")]
                + sorted_list_expr[(size_expr / 2).cast("long")]
            )
            / 2
        )
    )

使用方法如下：

output_df = input_spark_df.groupby("group").agg(
    median_exact("elems").alias("elems_median")
)

- J. Seegler

0

从版本 3.4+（以及已经在 3.3.1 中）开始中位数函数直接可用 https://github.com/apache/spark/blob/e170a2eb236a376b036730b5d63371e753f1d947/python/pyspark/sql/functions.py#L633

import pyspark.sql.functions as f

df.groupBy("grp").agg(f.median("val"))

如果版本最终发布，我猜相应的文档将会被添加。

- Jan_ewazz

0

在Spark中，计算数据框中的分位数（中位数、四分位数、百分位数等）可以精确或近似地完成。

示例数据框：

from pyspark.sql import functions as F
df = spark.range(1, 6)
df.show()
# +---+
# | id|
# +---+
# |  1|
# |  2|
# |  3|
# |  4|
# |  5|
# +---+

中位数：

准确，但对于大数据来说资源密集
- F.median('id') - 从Spark 3.4+版本开始
```
df.agg(F.median('id')).head()[0]
# 3.0
```
- F.expr('percentile(id, .5)')
```
df.agg(F.expr('percentile(id, .5)')).head()[0]
# 3.0
```
近似，但对于大数据来说高效
F.percentile_approx('id', .5)
```
df.agg(F.percentile_approx('id', .5)).head()[0]
# 3
```

下四分位数：

准确，但对于大数据资源消耗较多
F.expr('percentile(id, .25)')
```
df.agg(F.expr('percentile(id, .25)')).head()[0]
# 2.0
```
近似，但对于大数据效率高
F.percentile_approx('id', .25)
```
df.agg(F.percentile_approx('id', .25)).head()[0]
# 2
```

上四分位数：

准确，但对大数据资源密集型
F.expr('percentile(id, .75)')
```
df.agg(F.expr('percentile(id, .75)')).head()[0]
# 4.0
```
近似，但对大数据高效
F.percentile_approx('id', .75)
```
df.agg(F.percentile_approx('id', .75)).head()[0]
# 4
```

所有四分位数：

准确，但对大数据资源密集型
F.expr('percentile(id, array(.25, .5, .75))')

df.agg(F.expr('percentile(id, array(.25, .5, .75))')).head()[0]
# [2.0, 3.0, 4.0]

近似，但对大数据高效
F.percentile_approx('id', [.25, .5, .75])

df.agg(F.percentile_approx('id', [.25, .5, .75])).head()[0]
# [2, 3, 4]

其他分位数（例如百分位数）可以使用相同的方法进行计算（函数percentile或percentile_approx）。

- ZygD

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- zero323 · Accepted Answer

正在进行的工作

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

Spark 2.0+:

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

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

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

在我的几年旧的计算机上，大约需要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秒。可能有改进的方法，但为什么要费心呢？

语言无关（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 函数中，您可以传递一个额外的参数以确定要使用的记录数量。