Как найти квантили, используя Spark

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

Этот вопрос аналогичен этому вопросу. Тем не менее, ответ на вопрос – использование Scala, чего я не знаю.

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

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

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

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

РЕДАКТИРОВАТЬ:

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

2 Solutions collect form web for “Как найти квантили, используя Spark”

Spark 2.0+:

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

Python :

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

Скала :

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

где последний параметр является относительной ошибкой. Чем ниже число, тем точнее результаты и более дорогие вычисления.

Искры <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 

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

Добавление решения, если вы хотите только метод 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), eg 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) } 
  • Советы по правильному использованию больших широковещательных переменных?
  • Spark - Вложенная операция RDD
  • Ошибка при запуске Spark: Исполнитель проиграл
  • Как преобразовать Spark RDD в pandas dataframe в ipython?
  • Невозможно десериализовать RDD с различным количеством элементов в паре
  • Как удалить повторяющиеся значения из RDD
  • Spark dataframe преобразует несколько строк в столбец
  • Объект PipelinedRDD не имеет атрибута 'toDF' в PySpark
  • Какая функция в искре используется для объединения двух RDD по ключам
  • Spark ALS predAll возвращает пустое
  • Уменьшите пару «ключ-значение» в пару из списка ключей с помощью Apache Spark
  •  
    Interesting Posts for Van-Lav

    Как определить лазерную линию в изображении с помощью Python

    python или bash – добавление «в начале строки и», в конце строки

    Вставить код R в python

    Как фильтровать перестановки слов только для поиска семантически правильных nграмм? (Python 3, NLTK)

    Заменить весь текст между двумя строками python

    генерировать имена переменных на лету в python

    Python создает таблицу в переменной из файла csv

    Как jupyter может получить доступ к новому модулю tensorflow, установленному на правильном пути?

    Нет модуля с именем tkinter при использовании tkintertables

    Почему в itertools.permutations Python содержатся дубликаты? (Когда исходный список имеет дубликаты)

    распаковка файлов приводит к «BadZipFile: файл не является zip-файлом»

    Как складывать кадры данных друг на друга в Pandas

    Python: используйте ввод из другой команды

    Python: выход Dict элементов в производстве Corouts?

    Извлечь номер дома и название улицы из строки с использованием Python Regex

    Python - лучший язык программирования в мире.