Медианы/квантилей в группе PySparkBy

Я хотел бы рассчитать групповые квантили на кадре данных Spark (используя PySpark). Либо приблизительный или точный результат будет в порядке. Я предпочитаю решение, которое я могу использовать в контексте groupBy/agg, чтобы я мог смешать его с другими агрегатными функциями PySpark. Если по какой-то причине это невозможно, то подойдет и другой подход.

Этот вопрос связан, но не указывает, как использовать approxQuantile в качестве агрегатной функции.

У меня также есть доступ к UDF- percentile_approx Pertile Percent_approx, но я не знаю, как использовать его в качестве агрегатной функции.

Предположим, у меня есть следующий фрейм данных:

from pyspark import SparkContext
import pyspark.sql.functions as f

sc = SparkContext()    

df = sc.parallelize([
    ['A', 1],
    ['A', 2],
    ['A', 3],
    ['B', 4],
    ['B', 5],
    ['B', 6],
]).toDF(('grp', 'val'))

df_grp = df.groupBy('grp').agg(f.magic_percentile('val', 0.5).alias('med_val'))
df_grp.show()

Ожидаемый результат:

+----+-------+
| grp|med_val|
+----+-------+
|   A|      2|
|   B|      5|
+----+-------+

Ответ 1

Я думаю, тебе это больше не нужно. Но оставлю это здесь для будущих поколений (то есть меня на следующей неделе, когда я забуду).

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

grp_window = Window.partitionBy('grp')
magic_percentile = F.expr('percentile_approx(val, 0.5)')

df.withColumn('med_val', magic_percentile.over(grp_window))

Или, чтобы точно ответить на ваш вопрос, это также работает:

df.groupBy('gpr').agg(magic_percentile.alias('med_val'))

И в качестве бонуса вы можете передать массив процентилей:

quantiles = F.expr('percentile_approx(val, array(0.25, 0.5, 0.75))')

И вы получите список взамен.

Ответ 2

Поскольку у вас есть доступ к percentile_approx, одним простым решением было бы использовать его в команде SQL:

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)

df.registerTempTable("df")
df2 = sqlContext.sql("select grp, percentile_approx(val, 0.5) as med_val from df group by grp")

Ответ 3

К сожалению, и, насколько мне известно, кажется, что это невозможно сделать с помощью "чистых" команд PySpark (решение Shaido обеспечивает обходное решение с SQL), и причина очень элементарна: в отличие от с другими агрегатными функциями, такими как mean, approxQuantile не возвращает тип Column, а список.

Посмотрите на пример с примерами:

spark.version
# u'2.2.0'

import pyspark.sql.functions as func
from pyspark.sql import DataFrameStatFunctions as statFunc

# aggregate with mean works OK:
df_grp_mean = df.groupBy('grp').agg(func.mean(df['val']).alias('mean_val'))
df_grp_mean.show()
# +---+--------+ 
# |grp|mean_val|
# +---+--------+
# |  B|     5.0|
# |  A|     2.0|
# +---+--------+

# try aggregating by median:
df_grp_med = df.groupBy('grp').agg(statFunc(df).approxQuantile('val', [0.5], 0.1))
# AssertionError: all exprs should be Column

# mean aggregation is a Column, but median is a list:

type(func.mean(df['val']))
# pyspark.sql.column.Column

type(statFunc(df).approxQuantile('val', [0.5], 0.1))
# list

Я сомневаюсь, что подход на основе окон будет иметь какое-то значение, поскольку, как я сказал, основная причина очень элементарная.

См. также мой ответ здесь для получения более подробной информации.