云原生大数据计算服务 MaxCompute：Python 3 UDAF

• 导入模块：必选。

  至少要包含from odps.udf import annotate和from odps.udf import BaseUDAF。from odps.udf import annotate用于导入函数签名模块，MaxCompute才可以识别后续代码中定义的函数签名。from odps.udf import BaseUDAF为Python UDAF的基类，您需要通过此类在派生类中实现iterate、merge、terminate等方法。

  当UDAF代码中需要引用文件资源或表资源时，需要包含from odps.distcache import get_cache_file（文件资源）或from odps.distcache import get_cache_table（表资源）。

• 函数签名：必选。

  格式为@annotate(<signature>)，signature用于定义函数的输入参数和返回值的数据类型。更多函数签名信息，请参见函数签名及数据类型。

• 自定义Python类（派生类）：必选。

  UDAF代码的组织单位，定义了实现业务需求的变量及方法。您还可以在代码中引用MaxCompute内置的第三方库或引用文件、表资源。更多信息，请参见第三方库或引用资源。

• 实现Python类的方法：必选。

  Python类实现包含如下4个方法，您可以根据实际需要进行选择。

  方法定义	描述
  BaseUDAF.new_buffer()	返回聚合函数的中间值的buffer。buffer必须是Marshal对象（例如LIST、DICT），并且buffer的大小不应该随数据量递增。在极限情况下，buffer在执行对象序列化后的大小不应该超过2 MB。
  BaseUDAF.iterate(buffer[, args, ...])	将args聚合到中间值buffer中。
  BaseUDAF.merge(buffer, pbuffer)	将中间值buffer和pbuffer合并的结果存放在buffer中。
  BaseUDAF.terminate(buffer)	将buffer转换为MaxCompute SQL的基本类型。

#导入函数签名模块及基类。
from odps.udf import annotate
from odps.udf import BaseUDAF
#函数签名。
@annotate('double->double')
#自定义Python类。
class Average(BaseUDAF):
#实现Python类的方法。
    def new_buffer(self):
        return [0, 0]
    def iterate(self, buffer, number):
        if number is not None:
            buffer[0] += number
            buffer[1] += 1
    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
        buffer[1] += pbuffer[1]
    def terminate(self, buffer):
        if buffer[1] == 0:
            return 0.0
        return buffer[0] / buffer[1]

• 全新项目：新MaxCompute项目，或第一次使用Python语言编写UDAF的MaxCompute项目。建议所有的Python UDAF都直接使用Python 3语言编写。
• 存量项目：创建了大量Python 2 UDAF的MaxCompute项目。请您谨慎开启Python 3。如果您计划逐步将所有Python 2 UDAF迁移为Python 3 UDAF，推荐方法如下：
  • 新作业和新UDAF：使用Python 3语言编写，在Session级别开启Python 3。开启Python 3方法，请参见开启Python 3。
  • Python 2 UDAF：改写Python 2 UDAF，使其可以同时兼容Python 2和Python 3。改写方法请参见将Python 2代码移植到Python 3。
    说明 如果您需要编写公共UDAF，并为多个MaxCompute项目授权UDAF的操作权限，建议UDAF同时兼容Python 2和Python 3。

set odps.sql.python.version=cp37;

@annotate(<signature>)

'arg_type_list -> type'

• arg_type_list：表示输入参数的数据类型。输入参数可以为多个，用英文逗号（,）分隔。支持的数据类型为BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision,scale)、CHAR、VARCHAR、复杂数据类型（ARRAY、MAP、STRUCT）或复杂数据类型嵌套。
  arg_type_list还支持星号（*）或为空（''）：

  • 当arg_type_list为星号（*）时，表示输入参数为任意个数。
  • 当arg_type_list为空（''）时，表示无输入参数。
• type：表示返回值的数据类型。UDAF只返回一列。支持的数据类型为：BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision,scale)、复杂数据类型（ARRAY、MAP、STRUCT）或复杂数据类型嵌套。

• 在归属MaxCompute项目中使用自定义函数：使用方法与内建函数类似，您可以参照内建函数的使用方法使用自定义函数。
• 跨项目使用自定义函数：即在项目A中使用项目B的自定义函数，跨项目分享语句示例：select B:udf_in_other_project(arg0, arg1) as res from table_t;。更多跨项目分享信息，请参见基于Package跨项目访问资源。

• 您可以在参数列表中使用*，表示接受任意长度、任意类型的输入参数。例如@annotate('double,*->string')表示接受第一个参数是DOUBLE类型，后接任意长度、任意类型的参数列表。此时，您需要自己编写代码判断输入的个数和参数类型，然后对它们进行相应的操作（您可以对比C语言里面的printf函数来理解此操作）。
  说明 *用在返回值列表中时，表示的是不同的含义。
• UDAF的返回值可以使用*，表示返回任意个STRING类型。返回值的个数与调用函数时设置的别名个数有关。例如@annotate("bigint,string->double,*")，调用方式是UDTF(x, y) as (a, b, c)，此处as后面设置了三个别名，即a、b、c。编辑器会认定a为DOUBLE类型（Annotation中返回值第一列的类型是给定的），b和c为STRING类型。因为这里给出了三个返回值，所以UDTF在调用forward时，forward必须是长度为3的数组，否则会出现运行时报错。
  说明 这种错误无法在编译时报出，因此UDTF的调用者在SQL中设置alias个数时，必须遵循该UDAF定义的规则。由于聚合函数的返回值个数固定是1，所以这个功能对UDAF来说并无意义。

from odps.udf import annotate
from odps.udf import BaseUDAF
@annotate('bigint,*->string')
class MultiColSum(BaseUDAF):
    def new_buffer(self):
        return [0]
    def iterate(self, buffer, *args):
        for arg in args:
            buffer[0] += int(arg)
    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
    def terminate(self, buffer):
        return str(buffer[0])

-- 根据输入多个参数求和
SELECT my_multi_col_sum(a,b,c,d,e) from values (1,"2","3","4","5"), (6,"7","8","9","10") t(a,b,c,d,e);
-- 返回值为 55