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

PyODPS DataFrame admite varias operaciones de agregación en conjuntos de datos grandes: funciones de agregación integradas, agregaciones agrupadas, agregaciones personalizadas (funciones de agregación definidas por el usuario o UDAF) y estimación de valores distintos basada en HyperLogLog.

Todos los ejemplos utilizan la tabla pyodps_iris como origen de datos:

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

Funciones de agregación integradas

Las siguientes funciones de agregación están disponibles en las columnas de DataFrame.

Función	Descripción
`count` o `size`	Cuenta el número de filas
`unique`	Cuenta el número de valores distintos
`min`	Devuelve el valor mínimo
`max`	Devuelve el valor máximo
`sum`	Devuelve la suma total
`mean`	Devuelve el valor promedio
`median`	Devuelve la mediana
`quantile(p)`	Devuelve el cuantil p; solo devuelve resultados precisos para enteros
`var`	Devuelve la varianza
`std`	Devuelve la desviación estándar
`moment`	Devuelve el momento central N-ésimo o el momento N-ésimo
`skew`	Devuelve la asimetría muestral (estimación insesgada)
`kurtosis`	Devuelve la curtosis muestral (estimación insesgada)
`cat`	Concatena cadenas con un separador
`tolist`	Agrega una columna en una lista

Los DataFrames de PyODPS ignoran los valores nulos en las operaciones de agregación tanto en los backends de MaxCompute como en los de pandas. Este comportamiento difiere del de los DataFrames de pandas, pero coincide con la semántica de SQL.

Ejemplos

Describir todas las columnas numéricas: llame a describe() para obtener el recuento, el máximo, el mínimo, el promedio y la desviación estándar al mismo tiempo:

print(iris.describe())

Salida:

    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

Agregar una sola columna:

iris.sepallength.max()

Salida: 7.9

Agregar sobre valores distintos: llame a unique() antes de la función de agregación:

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

Salida: u'Iris-setosa,Iris-versicolor,Iris-virginica'

Agregar todas las columnas: si todas las columnas admiten la misma operación, aplíquela a todo el DataFrame:

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

Salida:

   sepal_length  sepal_width  petal_length  petal_width
1      5.843333     3.054000      3.758667     1.198667

Contar todas las filas:

iris.count()

Salida: 150

Para mostrar el resultado en los registros, ejecute print(iris.count().execute()) .

Agrupar y agregar datos

La operación groupby en PyODPS DataFrame sigue un modelo de dividir-aplicar-combinar:

Dividir: groupby() divide los datos en grupos según una o más columnas.
Aplicar: agg() o aggregate() aplica una función de agregación a cada grupo de forma independiente.
Combinar: los resultados se combinan en un solo DataFrame.

El resultado incluye tanto la columna agrupada como la columna agregada.

Agregación con nombre

Pase argumentos de palabra clave a agg() para controlar directamente los nombres de las columnas de salida. Esto se conoce como agregación con nombre; utilícela cuando necesite control explícito sobre los nombres de las columnas agregadas:

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

Salida:

              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

En este ejemplo, smin=iris.sepallength.min() cambia el nombre de la columna agregada a smin.

Contar valores distintos por grupo

Existen dos métodos equivalentes. Utilice value_counts() para mayor concisión:

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

Ambos producen la misma salida:

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

Agregar una sola columna de un grupo

Acceda a la columna por su nombre después de groupby() para recuperar únicamente esa columna agregada:

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

Salida:

   petallength_sum
0             73.2
1            213.0
2            277.6

Al usar esta sintaxis, está limitado a las funciones de agregación de esa columna. Para aplicar comprobaciones de valores no nulos u otras expresiones, use agg() en su lugar:

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

Salida:

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

Agrupar por un valor constante

Para agregar todas las filas juntas sin una columna de agrupación natural, agrupe por una constante mediante Scalar:

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

Salida:

   petallength_sum
0            563.8

Escribir agregaciones personalizadas

Utilice agg() o aggregate() para aplicar una función de agregación definida por el usuario (UDAF) a una columna. Una clase de agregación personalizada debe implementar cuatro métodos:

Método	Descripción
`buffer()`	Devuelve un objeto mutable (lista o diccionario) que acumula resultados parciales. El tamaño del búfer no debe aumentar con la cantidad de datos.
`__call__(buffer, *val)`	Agrega un valor al búfer.
`merge(buffer, pbuffer)`	Combina un búfer parcial (`pbuffer`) con el búfer principal.
`getvalue(buffer)`	Devuelve el valor agregado final.

Ejemplo: promedio personalizado

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)

Salida: 3.0540000000000007

Notas de uso

Especifique el tipo de datos de salida cuando sea diferente al de entrada:

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

Combinar con groupby:

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

Salida:

   petallength_aggregation
0                    3.418
1                    2.770
2                    2.974

Agregar varias columnas: use agg() de odps.df y pase una lista de columnas:

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)

Salida:

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

Llamar a una UDAF existente de MaxCompute por nombre: pase el nombre de la UDAF como cadena en lugar de una clase:

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

Debido a las limitaciones de las UDF de Python, los tipos LIST y DICT no pueden usarse como tipo de datos de entrada o salida para agregaciones personalizadas.

Conteo con HyperLogLog

hll_count implementa el algoritmo HyperLogLog para estimar el número de valores distintos en una columna. Devuelve un recuento aproximado, no exacto. Utilícelo cuando los recuentos exactos sean demasiado lentos en conjuntos de datos grandes; por ejemplo, para estimar visitantes únicos (UV) rápidamente.

El siguiente ejemplo utiliza un DataFrame de pandas. Ejecútelo en un entorno local. Si lo ejecuta en DataWorks, importe pandas primero mediante un paquete de terceros.

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

Salida: 63270

Como comparación, el recuento exacto de valores distintos:

df.a.nunique()

Salida: 63250

Los dos resultados son cercanos pero no idénticos: hll_count sacrifica un pequeño margen de error a cambio de velocidad.

Utilice el parámetro splitter para dividir valores de cadena antes de contar elementos distintos.