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

您可以通过MaxCompute Studio工具使用Python 3语言编写UDTF代码，代码中需要包含如下信息：

• 导入模块：必选。

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

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

• 函数签名：可选。

  格式为@annotate(<signature>)，signature用于定义函数的输入参数和返回值的数据类型。如果不指定函数签名，在SQL中调用UDTF时，可以匹配任意类型的输入参数，但返回值类型无法推导，所有输出参数都将会是STRING类型。更多函数签名信息，请参见函数签名及数据类型。

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

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

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

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

  方法定义	描述
  BaseUDTF.init()	初始化方法。派生类如果需要实现此方法，必须在一开始调用基类的初始化方法super(BaseUDTF, self).init()。init方法在整个UDTF生命周期中只会被调用一次，即在处理第一条记录之前。当UDTF需要保存内部状态时，可以通过此方法初始化所有状态。
  BaseUDTF.process([args, ...])	SQL中每一条记录都会对应调用一次process，process的参数为SQL语句中指定的UDTF输入参数。
  BaseUDTF.forward([args, ...])	UDTF的输出方法。此方法由用户代码调用。每调用一次forward，就会输出一条记录。forward的参数为SQL语句中指定的UDTF的输出参数。 如果Python代码中未指定函数签名，在调用forward方法时，必须将所有输出值转换为STRING类型。
  BaseUDTF.close()	UDTF的结束方法。只会被调用一次，即在处理完最后一条记录之后被调用。

UDTF代码示例如下。

#导入函数签名模块及基类。
from odps.udf import annotate
from odps.udf import BaseUDTF
#函数签名。
@annotate('string -> string')
#自定义Python类。
class Explode(BaseUDTF):
#实现Python类的方法。
   def process(self, arg):
       props = arg.split(',')
       for p in props:
           self.forward(p)

Python 2官方即将停止维护，建议您根据项目类型执行迁移操作：

• 全新项目：新MaxCompute项目，或第一次使用Python语言编写UDTF的MaxCompute项目。建议所有的Python UDTF都直接使用Python 3语言编写。

• 存量项目：创建了大量Python 2 UDTF的MaxCompute项目。请您谨慎开启Python 3。如果您计划逐步将所有Python 2 UDTF迁移为Python 3 UDTF，推荐方法如下：

  • 新作业和新UDTF：使用Python 3语言编写，在Session级别开启Python 3。开启Python 3方法，请参见开启Python 3。

  • Python 2 UDTF：改写Python 2 UDTF，使其可以同时兼容Python 2和Python 3。改写方法请参见将Python 2代码移植到Python 3。

    说明

    如果您需要编写公共UDTF，并为多个MaxCompute项目授权UDTF的操作权限，建议UDTF同时兼容Python 2和Python 3。

函数签名格式如下。

@annotate(<signature>)

signature为函数签名字符串，用于标识输入参数和返回值的数据类型。执行UDTF时，UDTF函数的输入参数和返回值类型要与函数签名指定的类型一致。查询语义解析阶段会检查不符合函数签名定义的用法，检查到类型不匹配时会报错。具体格式如下。

'arg_type_list -> type_list'

其中：

• type_list：表示返回值的数据类型。UDTF可以返回多列。支持的数据类型为：BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision,scale)、复杂数据类型（ARRAY、MAP、STRUCT）或复杂数据类型嵌套。

• 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为空（''）时，表示无输入参数。

  更多Resolve注解语法扩展详情，请参见UDAF和UDTF动态参数说明。

合法的函数签名示例如下。

引用文件资源和表资源的代码示例如下。

from odps.udf import annotate
from odps.udf import BaseUDTF
from odps.distcache import get_cache_file
from odps.distcache import get_cache_table
@annotate('string -> string, bigint')
class UDTFExample(BaseUDTF):
    """读取资源文件和资源表里的pageid、adid_list，生成dict
    """
    def __init__(self):
        import json
        cache_file = get_cache_file('test_json.txt')
        self.my_dict = json.load(cache_file)
        cache_file.close()
        records = list(get_cache_table('table_resource1'))
        for record in records:
            self.my_dict[record[0]] = record[1]
    """输入pageid，输出pageid以及它对应的所有adid
    """
    def process(self, pageid):
        for adid in self.my_dict[pageid]:
            self.forward(pageid, adid)

按照开发流程，完成Python 3 UDTF开发后，您即可通过MaxCompute SQL调用Python 3 UDTF。调用方法如下：

• 在归属MaxCompute项目中使用自定义函数：使用方法与内建函数类似，您可以参照内建函数的使用方法使用自定义函数。

• 跨项目使用自定义函数：即在项目A中使用项目B的自定义函数，跨项目分享语句示例：select B:udf_in_other_project(arg0, arg1) as res from table_t;。更多跨项目分享信息，请参见基于Package跨项目访问资源。