Médiane / quantiles dans PySpark groupBy

Je voudrais calculer groupe quantiles sur une Étincelle dataframe (à l'aide de PySpark). Soit un approximatif, ou le résultat exact ce serait bien. Je préfère une solution que je peux utiliser dans le cadre de groupBy /agg, de sorte que je peux le mélanger avec d'autres PySpark des fonctions d'agrégation. Si ce n'est pas possible pour une raison quelconque, une autre approche pourrait être très bien.

Cette question est lié, mais n'indique pas comment utiliser approxQuantile comme une fonction d'agrégation.

J'ai aussi accès à la percentile_approx Ruche de l'UDF, mais je ne sais pas comment l'utiliser comme une fonction d'agrégation.

Par souci de simplicité, supposons que j'ai la suite dataframe:

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()

Résultat attendu est:

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

Je pense que vous pourriez être en mesure de rouler dans cette instance à l'aide de la sous-jacentes rdd et un algorithme pour le calcul distribué quantiles par exemple, ici et les liens qui y sont. En fait, le github rapport a certains pyspark exemples.
Comment utiliser approxQuantile par groupe?

OriginalL'auteur abeboparebop | 2017-10-20

Je suppose que vous en avez le plus besoin. Mais le laisser ici pour les générations futures (c'est à dire moi la semaine prochaine si j'oublie).

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

Ou pour répondre exactement à votre question, cela fonctionne aussi:

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

Et comme un bonus, vous pouvez passer un tableau de percentiles:

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

Et vous obtiendrez une liste des en retour.

OriginalL'auteur kael

10

Puisque vous avez accès à percentile_approx, une solution simple serait d'utiliser dans un SQL commande:
```
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")
```
Cela fonctionne, mais je préfère une solution que je peux utiliser dans groupBy / agg à la PySpark niveau (alors que je peux facilement mélanger avec d'autres PySpark fonctions d'agrégation).
Je ne crois qu'il est possible d'utiliser seulement groupBy et agg, toutefois, pour utiliser une fenêtre de base de l'approche devrait également fonctionner.
J'ai clarifié ma solution idéale dans la question. Clairement cette réponse fait le travail, mais ce n'est pas tout à fait ce que je veux. Je vais laisser la question ouverte pour quelques temps pour voir si un nettoyeur de réponse.

OriginalL'auteur Shaido

Malheureusement, et au meilleur de ma connaissance, il semble qu'il n'est pas possible de le faire avec "pure" PySpark commandes (la solution par Shaido fournit une solution de contournement avec SQL), et la raison en est très simple: en contraste avec les autres fonctions d'agrégation, comme mean, approxQuantile ne renvoie pas un Column type, mais un liste.

Nous allons voir un exemple rapide avec vos données d'échantillonnage:

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

Je doute qu'une fenêtre de l'approche fondée sur le fait aucune différence, puisque comme je l'ai dit la raison sous-jacente est très élémentaire.

Voir aussi ma réponse ici pour plus de détails.

OriginalL'auteur desertnaut

Vous devez vous connecter pour publier un commentaire.