Как найти медианные и квантильные числа, используя Spark

24

Как я могу найти медиану из RDD целых чисел, используя распределенный метод, IPython и Spark? % Co_de% составляет приблизительно 700 000 элементов и поэтому слишком велико для сбора и поиска медианы.

Этот вопрос похож на этот вопрос. Однако ответ на вопрос - использование Scala, чего я не знаю.

Как вычислить точную медиану с Apache Spark?

Используя мышление для ответа Scala, я пытаюсь написать аналогичный ответ на Python.

Я знаю, что сначала хочу отсортировать RDD . Я не знаю как. Я вижу RDD (Сортирует этот RDD по данным sortBy ) и keyfunc (Сортирует этот sortByKey , который предположительно состоит из пар (ключ, значение).). Я думаю, что оба используют ключевое значение, а мой RDD имеет только целые элементы.

  1. Во-первых, я думал сделать RDD ?
  2. Далее я найду длину rdd ( myrdd.sortBy(lambda x: x) ).
  3. Наконец, я хочу найти элемент или 2 элемента в центре rdd. Мне также нужна помощь в этом методе.

EDIT:

У меня была идея. Возможно, я могу проиндексировать мой rdd.count() , а затем key = index и value = element. И тогда я могу попытаться сортировать по значению? Я не знаю, возможно ли это, потому что существует только метод RDD .

    
задан pr338 15.07.2015 в 16:11
источник
  • Ну, с целыми числами 7e5, предполагающими 64 бита каждый, вам нужно около 5 МБ для хранения всех данных, и для вычисления медианного локального значения требуется доля секунды, используя np.median :) Конечно, вы можете сортировать и индексировать, как описано но я предполагаю, что он будет примерно на порядок медленнее. –  zero323 15.07.2015 в 18:42
  • zero323: Возможно, это кластер Spark, работающий на кластере Commodore 64s. –  Daniel Darabos 15.07.2015 в 19:05
  • @ DanielDarabos Это злая идея :) И кассетные деки, как замена HDFS ... –  zero323 15.07.2015 в 19:37
  • Вот как это сделать с помощью Pyspark Dataframe AP: stackoverflow.com/questions/38743476/... –  gench 03.08.2016 в 16:13

3 ответа

48

Spark 2.0 +:

Вы можете использовать метод approxQuantile , который реализует алгоритм Greenwald-Khanna :

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)

Spark & ​​lt; 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,5 МБ памяти.

Если данные намного больше, сортировка будет ограничивающим фактором, поэтому вместо получения точного значения, вероятно, лучше попробовать, собрать и вычислить локально. Но если вы действительно хотите использовать 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 , вы также можете использовать UDAF для Hive. Со встроенными значениями:

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 вы можете передать дополнительный аргумент, который определяет количество используемых записей.

    
ответ дан zero323 15.07.2015 в 19:30
источник
  • Можно ли в Spark 2.0 использовать функцию approxQuantile () с функциями окна? Например, если необходимо вычислить движущуюся медиану на DataFrame. –  user3791111 10.06.2016 в 18:16
  • @ user3791111 Маловероятно, и в этом не было бы никакой ценности. Когда вы используете функции окна, вы можете получить точное значение в окне без каких-либо дополнительных затрат. –  zero323 10.06.2016 в 19:20
  • ОК, точный или приблизительный - независимо от того, будет ли какой-либо способ рассчитать «движущуюся медианную» (НЕ «скользящую среднюю») в Spark 2.0? –  user3791111 13.06.2016 в 13:00
5

Добавление решения, если вы хотите только метод RDD и не хотите переходить в DF. Этот фрагмент может дать вам процентиль для двойной двойной копии.

Если вы вводите процентиль как 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 09.09.2016 в 00:18
источник
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")
    
ответ дан Benoît Carne 09.11.2017 в 16:19
источник
  • должен ли я добавлять что-то еще с тех пор, как я попробовал его, и NameError: имя «DataFrame» не определено. –  E B 20.04.2018 в 00:02
  • Вы правы, импорт отсутствовал. Я соответствующим образом обновил ответ. благодаря –  Benoît Carne 26.04.2018 в 12:39