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

Question

如何RDD使用分布式方法,IPython和Spark 找到整数的中位数？的RDD是约700 000元,因此过大,以收集和发现中位数.

这个问题与这个问题类似.但是,问题的答案是使用Scala,我不知道.

如何使用Apache Spark计算精确中位数？

使用Scala答案的思考,我试图在Python中编写类似的答案.

我知道我首先要排序RDD.我不知道怎么.我看到sortBy(按给定的方式对此RDD进行排序keyfunc)和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 2.0+:

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

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)

Spark <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

我几年前的电脑和大约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你可以通过它决定了多项纪录使用一个额外的参数.

Answer 2

如果您只想要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)
  }

Answer 3

这是我使用窗口函数(使用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")