阿里云云原生大数据计算服务 MaxCompute聚合操作-云淘科技

分组聚合

分组聚合操作如下：

• DataFrame提供了groupby函数执行分组操作，分组后通过调用agg或者aggregate方法，执行聚合操作。最终的结果列中会包含分组的列和聚合的列。

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

  返回结果如下。

                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

• DataFrame提供了value_counts函数，按某列分组后，将每个组的个数从大到小进行排列。

  • 使用groupby函数实现。

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

    返回结果如下。

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

  • 使用value_counts函数实现。

    iris['name'].value_counts().head(5)

    返回结果如下。

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

• 对于聚合后的单列操作，您也可以直接取出列名。但此时只能使用聚合函数。

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

  返回结果如下。

     petallength_sum
  0             73.2
  1            213.0
  2            277.6

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

  返回结果如下。

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

• 分组时也支持对常量进行分组，但是需要使用Scalar初始化。

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

  返回结果如下。

     petallength_sum
  0            563.8

编写自定义聚合

对字段使用agg或者aggregate方法调用自定义聚合。自定义聚合需要提供一个类，这个类需要提供以下方法：

• buffer()：返回一个Mutable的Object（例如LIST或DICT），buffer大小不应随数据量增大而递增。

• __call__(buffer, *val)：将值聚合到中间buffer。

• merge(buffer, pbuffer)：将pbuffer聚合到buffer中。

• getvalue(buffer)：返回最终值。

计算平均值的示例如下。

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)

返回结果如下。

3.0540000000000007

编写自定义聚合还需要关注如下内容：

• 如果最终类型和输入类型发生了变化，则需要指定类型。

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

• 自定义聚合也可以用在分组聚合中。

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

  返回结果如下。

     petallength_aggregation
  0                    3.418
  1                    2.770
  2                    2.974

• 对多列可以使用agg方法调用自定义聚合。

  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, iris.sepallength], Agg, rtype='float')  # 对两列调用自定义聚合。
  iris.groupby('name').agg(val=to_agg)

  返回结果如下。

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

• 如果您需要调用MaxCompute上已经存在的UDAF，指定函数名即可。

  iris.groupby('name').agg(iris.sepalwidth.agg('your_func'))  # 对单列聚合。
  to_agg = agg([iris.sepalwidth, iris.sepallength], 'your_func', rtype='float')
  iris.groupby('name').agg(to_agg.rename('val'))  # 对多列聚合。

  说明

  目前，因受限于Python UDF，自定义聚合无法支持将LIST或DICT类型作为初始输入或最终输出结果。

HyperLogLog计数

DataFrame提供了对列进行HyperLogLog计数的接口hll_count，这个接口是近似个数的估计接口。当数据量很大时，它可以较快地估计去重后的数据量。

使用该接口计算海量用户UV时，可以快速得出估计值。

说明

以下示例使用了Pandas包，您可以在本地环境中运行以下示例，如果在DataWorks环境中运行，您需要先通过三方包的方式导入Pandas包。

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

返回结果如下。

63270

df.a.nunique()

返回结果如下。

63250

说明

splitter参数会对每个字段进行分隔，再计算去重后的数据量。