How to perform grouping and aggregation and write a custom aggregation - MaxCompute: Central de Documentação do Alibaba Cloud

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

Funções de agregação integradas

As seguintes funções de agregação estão disponíveis nas colunas do DataFrame.

Função	Descrição
`count` ou `size`	Conta o número de linhas
`unique`	Conta o número de valores distintos
`min`	Retorna o valor mínimo
`max`	Retorna o valor máximo
`sum`	Retorna a soma total
`mean`	Retorna o valor médio
`median`	Retorna o valor mediano
`quantile(p)`	Retorna o p-quantil; retorna resultados precisos apenas para números inteiros
`var`	Retorna a variância
`std`	Retorna o desvio padrão
`moment`	Retorna o N-ésimo momento central ou o N-ésimo momento
`skew`	Retorna a assimetria amostral (estimativa não tendenciosa)
`kurtosis`	Retorna a curtose amostral (estimativa não tendenciosa)
`cat`	Concatena strings com um separador
`tolist`	Agrega uma coluna em uma lista

Os DataFrames do PyODPS ignoram valores nulos nas operações de agregação em ambos os backends MaxCompute e pandas. Esse comportamento difere do DataFrame do pandas, mas corresponde à semântica SQL.

Exemplos

Descreva todas as colunas numéricas — chame describe() para obter de uma vez o count, max, min, mean e desvio padrão:

print(iris.describe())

Saída:

    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 uma única coluna:

iris.sepallength.max()

Saída: 7.9

Agregar sobre valores distintos — chame unique() antes da função de agregação:

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

Saída: u'Iris-setosa,Iris-versicolor,Iris-virginica'

Agregar todas as colunas — se todas as colunas suportam a mesma operação, aplique-a ao DataFrame inteiro:

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

Saída:

   sepal_length  sepal_width  petal_length  petal_width
1      5.843333     3.054000      3.758667     1.198667

Contar todas as linhas:

iris.count()

Saída: 150

Para exibir o resultado nos logs, execute print(iris.count().execute()) .

Agrupar e agregar dados

O Groupby no PyODPS DataFrame segue um modelo split-apply-combine:

Split: groupby() divide os dados em grupos com base em uma ou mais colunas.
Apply: agg() ou aggregate() aplica uma função de agregação a cada grupo de forma independente.
Combine: os resultados são combinados em um único DataFrame.

O resultado inclui tanto a coluna agrupada quanto a coluna agregada.

Agregação nomeada

Passe argumentos de palavra-chave para agg() para controlar diretamente os nomes das colunas de saída. Isso é a agregação nomeada — use-a quando precisar de controle explícito sobre os nomes das colunas agregadas:

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

Saída:

              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

Neste exemplo, smin=iris.sepallength.min() renomeia a coluna agregada para smin.

Contar valores distintos por grupo

Há duas abordagens equivalentes disponíveis. Use value_counts() pela concisão:

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

Ambas produzem a mesma saída:

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

Agregar uma única coluna de um grupo

Acesse a coluna pelo nome após groupby() para recuperar apenas essa coluna agregada:

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

Saída:

   petallength_sum
0             73.2
1            213.0
2            277.6

Ao usar essa sintaxe, você está limitado às funções de agregação nessa coluna. Para aplicar verificações de não nulo ou outras expressões, use agg() em vez disso:

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

Saída:

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

Agrupar por um valor constante

Para agregar todas as linhas juntas sem uma coluna de agrupamento natural, agrupe por uma constante usando Scalar:

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

Saída:

   petallength_sum
0            563.8

Criar agregações personalizadas

Use agg() ou aggregate() para aplicar uma função de agregação definida pelo usuário (UDAF) a uma coluna. Uma classe de agregação personalizada deve implementar quatro métodos:

Método	Descrição
`buffer()`	Retorna um objeto mutável (lista ou dict) que acumula resultados parciais. O tamanho do buffer não deve crescer com a quantidade de dados.
`__call__(buffer, *val)`	Adiciona um valor ao buffer.
`merge(buffer, pbuffer)`	Mescla um buffer parcial (`pbuffer`) no buffer principal.
`getvalue(buffer)`	Retorna o valor final agregado.

Exemplo: média personalizada

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)

Saída: 3.0540000000000007

Notas de uso

Especifique o tipo de dado de saída quando ele diferir do tipo de entrada:

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

Combinação com groupby:

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

Saída:

   petallength_aggregation
0                    3.418
1                    2.770
2                    2.974

Agregar várias colunas — use agg() de odps.df e passe uma lista de colunas:

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)

Saída:

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

Chamar uma UDAF existente do MaxCompute pelo nome — passe o nome da UDAF como string em vez de uma 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'))

Devido a limitações do Python UDF, os tipos LIST e DICT não podem ser usados como tipo de dado de entrada ou saída para agregações personalizadas.

Contagem com HyperLogLog

hll_count implementa o algoritmo HyperLogLog para estimar o número de valores distintos em uma coluna. Ele retorna uma contagem aproximada, não exata. Use-o quando contagens exatas forem muito lentas em grandes conjuntos de dados — por exemplo, para estimar rapidamente visitantes únicos (UVs).

O exemplo abaixo usa um DataFrame do pandas. Execute-o em um ambiente local. Se estiver executando no DataWorks, importe o pandas via pacote de terceiros primeiro.

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

Saída: 63270

Para comparação, a contagem exata de valores distintos:

df.a.nunique()

Saída: 63250

Os dois resultados são próximos, mas não idênticos — hll_count troca uma pequena margem de erro por velocidade.

Use o parâmetro splitter para dividir valores de string antes de contar elementos distintos.