How to perform grouping and aggregation and write a custom aggregation - MaxCompute

from odps.df import DataFrame
iris = DataFrame(o.get_table('pyodps_iris'))

Fonctions d'agrégation intégrées

Les fonctions d'agrégation suivantes sont disponibles pour les colonnes DataFrame.

Fonction	Description
`count` ou `size`	Compte le nombre de lignes
`unique`	Compte le nombre de valeurs distinctes
`min`	Renvoie la valeur minimale
`max`	Renvoie la valeur maximale
`sum`	Renvoie la somme totale
`mean`	Renvoie la valeur moyenne
`median`	Renvoie la valeur médiane
`quantile(p)`	Renvoie le quantile p ; renvoie des résultats exacts uniquement pour les entiers
`var`	Renvoie la variance
`std`	Renvoie l'écart type
`moment`	Renvoie le moment centré d'ordre N ou le moment d'ordre N
`skew`	Renvoie l'asymétrie de l'échantillon (estimation sans biais)
`kurtosis`	Renvoie l'aplatissement de l'échantillon (estimation sans biais)
`cat`	Concatène les chaînes avec un séparateur
`tolist`	Agrège une colonne dans une liste

PyODPS DataFrames ignorent les valeurs nulles lors des opérations d'agrégation, que ce soit avec les backends MaxCompute ou pandas. Ce comportement diffère de celui des DataFrame pandas, mais correspond à la sémantique SQL.

Exemples

Décrire toutes les colonnes numériques — appelez describe() pour obtenir simultanément le compte, le maximum, le minimum, la moyenne et l'écart type :

print(iris.describe())

Résultat :

    type  sepal_length  sepal_width  petal_length  petal_width
0  count    150.000000   150.000000    150.000000   150.000000
1   mean      5.843333     3.054000      3.758667     1.198667
2    std      0.828066     0.433594      1.764420     0.763161
3    min      4.300000     2.000000      1.000000     0.100000
4    max      7.900000     4.400000      6.900000     2.500000

Agréger une seule colonne :

iris.sepallength.max()

Résultat : 7.9

Agréger sur les valeurs distinctes — appelez unique() avant la fonction d'agrégation :

iris.name.unique().cat(sep=',')

Résultat : u'Iris-setosa,Iris-versicolor,Iris-virginica'

Agréger toutes les colonnes — si toutes les colonnes prennent en charge la même opération, appliquez-la à l'intégralité du DataFrame :

iris.exclude('category').mean()

Résultat :

   sepal_length  sepal_width  petal_length  petal_width
1      5.843333     3.054000      3.758667     1.198667

Compter toutes les lignes :

iris.count()

Résultat : 150

Pour afficher le résultat dans les journaux, exécutez print(iris.count().execute()) .

Regrouper et agréger des données

L'opération Groupby dans PyODPS DataFrame suit un modèle split-apply-combine :

Split : groupby() divise les données en groupes basés sur une ou plusieurs colonnes.
Apply : agg() ou aggregate() applique une fonction d'agrégation à chaque groupe de manière indépendante.
Combine : les résultats sont combinés en un seul DataFrame.

Le résultat inclut à la fois la colonne de regroupement et la colonne agrégée.

Agrégation nommée

Transmettez des arguments nommés à agg() pour contrôler directement les noms des colonnes de sortie. Il s'agit d'une agrégation nommée ; utilisez cette approche chaque fois que vous souhaitez contrôler explicitement les noms des colonnes agrégées :

iris.groupby('name').agg(iris.sepallength.max(), smin=iris.sepallength.min())

Résultat :

              name  sepallength_max  smin
0      Iris-setosa              5.8   4.3
1  Iris-versicolor              7.0   4.9
2   Iris-virginica              7.9   4.9

Dans cet exemple, smin=iris.sepallength.min() renomme la colonne agrégée en smin.

Compter les valeurs distinctes par groupe

Deux approches équivalentes sont disponibles. Utilisez value_counts() pour plus de concision :

# Using groupby + agg
iris.groupby('name').agg(count=iris.name.count()).sort('count', ascending=False).head(5)

# Using value_counts (equivalent, more concise)
iris['name'].value_counts().head(5)

Les deux méthodes produisent le même résultat :

              name  count
0   Iris-virginica     50
1  Iris-versicolor     50
2      Iris-setosa     50

Agréger une seule colonne d'un groupe

Accédez à la colonne par son nom après groupby() pour récupérer uniquement cette colonne agrégée :

iris.groupby('name').petallength.sum()

Résultat :

   petallength_sum
0             73.2
1            213.0
2            277.6

Cette syntaxe limite l'utilisation aux fonctions d'agrégation sur cette colonne. Pour appliquer des vérifications de non-nullité ou d'autres expressions, utilisez plutôt agg() :

iris.groupby('name').agg(iris.petallength.notnull().sum())

Résultat :

              name  petallength_sum
0      Iris-setosa               50
1  Iris-versicolor               50
2   Iris-virginica               50

Regrouper par une valeur constante

Pour agréger toutes les lignes ensemble sans colonne de regroupement naturelle, regroupez par une constante en utilisant Scalar :

from odps.df import Scalar
iris.groupby(Scalar(1)).petallength.sum()

Résultat :

   petallength_sum
0            563.8

Écrire des agrégations personnalisées

Utilisez agg() ou aggregate() pour appliquer une fonction d'agrégation définie par l'utilisateur (UDAF) à une colonne. Une classe d'agrégation personnalisée doit implémenter quatre méthodes :

Méthode	Description
`buffer()`	Renvoie un objet mutable (liste ou dictionnaire) qui accumule les résultats partiels. La taille du buffer ne doit pas augmenter avec la quantité de données.
`__call__(buffer, *val)`	Ajoute une valeur au buffer.
`merge(buffer, pbuffer)`	Fusionne un buffer partiel (`pbuffer`) dans le buffer principal.
`getvalue(buffer)`	Renvoie la valeur agrégée finale.

Exemple : moyenne personnalisée

class Agg(object):

    def buffer(self):
        return [0.0, 0]

    def __call__(self, buffer, val):
        buffer[0] += val
        buffer[1] += 1

    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
        buffer[1] += pbuffer[1]

    def getvalue(self, buffer):
        if buffer[1] == 0:
            return 0.0
        return buffer[0] / buffer[1]

iris.sepalwidth.agg(Agg)

Résultat : 3.0540000000000007

Notes d'utilisation

Spécifiez le type de données de sortie lorsqu'il diffère de l'entrée :

iris.sepalwidth.agg(Agg, 'float')

Combinez avec groupby :

iris.groupby('name').sepalwidth.agg(Agg)

Résultat :

   petallength_aggregation
0                    3.418
1                    2.770
2                    2.974

Agréger plusieurs colonnes — utilisez agg() depuis odps.df et transmettez une liste de colonnes :

class Agg(object):

    def buffer(self):
        return [0.0, 0.0]

    def __call__(self, buffer, val1, val2):
        buffer[0] += val1
        buffer[1] += val2

    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
        buffer[1] += pbuffer[1]

    def getvalue(self, buffer):
        if buffer[1] == 0:
            return 0.0
        return buffer[0] / buffer[1]

from odps.df import agg
to_agg = agg([iris.sepalwidth, ], Agg, rtype='float')  # Call a user-defined aggregate function (UDAF) to aggregate data in two columns.
iris.groupby('name').agg(val=to_agg)

Résultat :

              name       val
0      Iris-setosa  0.682781
1  Iris-versicolor  0.466644
2   Iris-virginica  0.451427

Appeler une UDAF MaxCompute existante par son nom — transmettez le nom de l'UDAF sous forme de chaîne plutôt qu'une classe :

# Single column
iris.groupby('name').agg(iris.sepalwidth.agg('your_func'))

# Multiple columns
to_agg = agg([iris.sepalwidth, ], 'your_func', rtype='float')
iris.groupby('name').agg(to_agg.rename('val'))

En raison des limitations des UDF Python, les types LIST et DICT ne peuvent pas être utilisés comme type de données d'entrée ou de sortie pour les agrégations personnalisées.

Comptage HyperLogLog

hll_count implémente l'algorithme HyperLogLog pour estimer le nombre de valeurs distinctes dans une colonne. Il renvoie un compte approximatif, et non exact. Utilisez-le lorsque les comptes exacts sont trop lents sur de grands ensembles de données, par exemple pour estimer rapidement les visiteurs uniques (UV).

L'exemple ci-dessous utilise un DataFrame pandas. Exécutez-le dans un environnement local. Si vous l'exécutez dans DataWorks, importez d'abord pandas via un package tiers.

from odps.df import DataFrame
import pandas as pd
import numpy as np

df = DataFrame(pd.DataFrame({'a': np.random.randint(100000, size=100000)}))
df.a.hll_count()

Résultat : 63270

À titre de comparaison, le compte distinct exact :

df.a.nunique()

Résultat : 63250

Les deux résultats sont proches mais pas identiques — hll_count sacrifie une petite marge de précision pour gagner en vitesse.

Utilisez le paramètre splitter pour fractionner les valeurs de chaîne avant de compter les éléments distincts.