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

【问题标题】：How to find median and quantiles using Spark如何使用 Spark 查找中位数和分位数
【发布时间】：2015-10-04 15:18:27
【问题描述】：

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

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

编辑：

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

【问题讨论】：

好吧，对于 7e5 个整数，假设每个 64 位，您需要大约 5MB 来存储所有数据，并且使用 np.median 在本地计算中位数需要几分之一秒 :) 当然，您可以排序和你描述的索引，但我猜它会慢一些并且数量级要慢。
zero323：可能是在 Commodore 64 集群上运行的 Spark 集群。
@DanielDarabos 这是一个邪恶的想法 :) 和磁带卡座作为 HDFS 的替代品......
这里是如何使用 Pyspark Dataframe AP：stackoverflow.com/questions/38743476/…

标签： python apache-spark median rdd pyspark

【解决方案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)

底层方法也可以使用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

火花

Python

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

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

【讨论】：

在 Spark 2.0 中是否可以将 approxQuantile() 与窗口函数一起使用？例如，如果需要计算 DataFrame 上的移动中位数。
@user3791111 不太可能，而且没有任何价值。当您使用窗口函数时，您可以在窗口中获得准确的值，而无需额外费用。
好的，精确的或近似的——无论如何，在 Spark 2.0 中是否有任何方法可以计算“移动中位数”（不是“移动平均线”）？

【解决方案2】：

这是我使用窗口函数的方法（使用 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")

【讨论】：

我是否应该添加其他内容，因为我尝试过并且 NameError: name 'DataFrame' is not defined ..
你是对的，导入丢失了。我相应地更新了答案。谢谢
这不适用于组中的偶数：中位数会很差。它必须是两个中间元素之间的平均值。
@BenoîtCarne 这条DataFrame.addMedian = addMedian 线路如何运作？在 Python 中叫什么？
@Shankar 不确定它是否有正式的 Python 名称！我将其称为“在定义后向类添加函数”。更多关于这里的信息：stackoverflow.com/questions/9455111/… 我这样做只是为了能够调用 addMedian 函数，就好像它已经在 Spark 中本地实现一样。这不是强制性的。

【解决方案3】：

如果您只需要 RDD 方法并且不想移动到 DF，请添加解决方案。对于双倍 RDD，这个 sn-p 可以获得一个百分位。

如果您输入百分位数为 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)
  }

【讨论】：

【解决方案4】：

我编写了一个函数，它将数据帧作为输入并返回一个数据帧，该数据帧在分区上具有中值作为输出，order_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")

【讨论】：

【解决方案5】：

有两种方法可以使用。一种是使用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.51)
)*.5).alias("MEDIAN))

【讨论】：

【解决方案6】：

对于精确的中值计算，您可以使用以下函数并将其与 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")
)

【讨论】：