Python 2 UDF

使用限制

基于上述原因，您上传的代码都必须通过标准Python实现，禁止C扩展模块。

第三方库

参数与返回值

@odps.udf.annotate(signature)

Python UDF支持的MaxCompute SQL数据类型包括BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、复杂数据类型（ARRAY、MAP和STRUCT）和复杂数据类型嵌套。在执行SQL语句前，必须确定所有函数的参数类型和返回值类型，因此对于Python这一动态类型语言，需要通过对UDF类加Decorator的方式指定函数签名。

arg_type_list '->' type_list
arg_type_list: type_list | '*' | '' | 'char(n)' | 'varchar(n)'
type_list: [type_list ','] type
type: 'bigint' | 'string' | 'double' | 'boolean' | 'datetime' | 'float' | 'binary' | 'date' | 'decimal' | 'decimal(precision,scale)'

'bigint,double->string'            # 参数类型为BIGINT、DOUBLE，返回值类型为STRING。
'bigint,boolean->string,datetime'  # UDTF参数类型为BIGINT、BOOLEAN，返回值类型为STRING，DATETIME。
'*->string'                        # 变长参数，输入参数类型任意，返回值类型为STRING。
'->double'                         # 参数为空，返回值类型为DOUBLE。
'array<bigint>->struct<x:string, y:int>' # 参数类型为ARRAY<BIGINT>，返回值类型为STRUCT<x:STRING, y:INT>。
'->map<bigint, string>'            # 参数为空，返回值类型为MAP<BIGINT, STRING>。

查询语义解析阶段会检查不符合函数签名的用法，并返回错误，禁止执行此函数。执行时，UDF函数的参数会以函数签名指定的类型传入。返回值类型也要与函数签名指定的类型一致，否则检查到类型不匹配时也会报错。

UDF

from odps.udf import annotate
@annotate("bigint,bigint->bigint")
class MyPlus(object):
   def evaluate(self, arg0, arg1):
       if None in (arg0, arg1):
           return None
       return arg0 + arg1

UDAF

• class odps.udf.BaseUDAF：继承此类实现Python UDAF。
• 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的基本类型。

@annotate('double->double')
class Average(BaseUDAF):
    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]

UDTF

• class odps.udf.BaseUDTF：Python UDTF的基类。您可以继承此类实现process或close等方法。
• BaseUDTF.__init__()：初始化方法。继承类如果需要实现此方法，必须在一开始调用基类的初始化方法super(BaseUDTF, self).__init__()。init方法在整个UDTF生命周期中只会被调用一次，即在处理第一条记录之前。当UDTF需要保存内部状态时，可以通过此方法初始化所有状态。
• BaseUDTF.process([args, ...])：此方法由MaxCompute SQL框架调用，SQL中每一条记录都会调用一次process，process的参数为SQL语句中指定的UDTF输入参数。
• BaseUDTF.forward([args, ...])：UDTF的输出方法。此方法由用户代码调用。每调用一次forward，便会输出一条记录。 forward的参数为SQL语句中指定的UDTF的输出参数。
• BaseUDTF.close()：UDTF的结束方法。此方法由MaxCompute SQL框架调用，并且只会被调用一次，即在处理完最后一条记录之后。

#coding:utf-8
# explode.py
from odps.udf import annotate
from odps.udf import BaseUDTF
@annotate('string -> string')
class Explode(BaseUDTF):
   #将string按逗号分隔输出成多条记录。
   def process(self, arg):
       props = arg.split(',')
       for p in props:
           self.forward(p)

引用资源

Python UDF可以通过odps.distcache模块引用资源文件，支持引用文件资源和表资源。

• odps.distcache.get_cache_file(resource_name)：返回指定名字的资源内容。
  • resource_name为STRING类型，对应当前项目空间中已存在的资源名。如果资源名非法或者没有相应的资源，则会返回异常。
    说明 使用UDF访问资源，在创建UDF时需要声明引用的资源，否则会报错。
  • 返回值为File-like对象。在使用完此对象后，您需要调用close方法释放打开的资源文件。
  示例如下。

  @annotate('bigint->string')
  class DistCacheExample(object):
  def __init__(self):
      cache_file = get_cache_file('test_distcache.txt')
      kv = {}
      for line in cache_file:
          line = line.strip()
          if not line:
              continue
          k, v = line.split()
          kv[int(k)] = v
      cache_file.close()
      self.kv = kv
  def evaluate(self, arg):
      return self.kv.get(arg)

• odps.distcache.get_cache_table(resource_name)：返回指定资源表的内容。
  • resource_name支持STRING类型，对应当前Project中已存在的资源表名。如果资源名非法或者没有相应的资源，会返回异常。
  • 返回值为GENERATOR类型，调用者通过遍历获取表的内容，每次遍历可得到以数组形式存在的表中的一条记录。
  示例如下。

  from odps.udf import annotate
  from odps.distcache import get_cache_table
  @annotate('->string')
  class DistCacheTableExample(object):
      def __init__(self):
          self.records = list(get_cache_table('udf_test'))
          self.counter = 0
          self.ln = len(self.records)
      def evaluate(self):
          if self.counter > self.ln - 1:
              return None
          ret = self.records[self.counter]
          self.counter += 1
          return str(ret)

UDTF读取MaxCompute资源示例

在Python UDTF中，您可以读取MaxCompute的资源。利用UDTF读取MaxCompute资源的操作流程示例如下：

1. 编写Python代码，并保存为py_udtf_example.py文件。代码示例如下：

   # -*- coding: utf-8 -*-
   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，生成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)

2. 准备资源文件test_json.txt、创建资源表table_resource1和内部表tmp1（后续执行DML操作写入的目标表）并插入数据。命令示例如下：
   • 资源文件test_json.txt的内容如下：

     {"front_page":[1, 2, 3], "contact_page1":[3, 4, 5]}

   • 创建资源表table_resource1，并插入数据。

     create table if not exists table_resource1 (pageid string, adid int);
     insert into table table_resource1 values("contact_page2",2),("contact_page3",5);

   • 创建内部表tmp1，并插入数据。

     create table if not exists tmp1 (pageid string);
     insert into table tmp1 values ("front_page"),("contact_page1"),("contact_page3");

3. 登录MaxCompute客户端将py_udtf_example.py文件、test_json.txt和表table_resource1添加为MaxCompute的资源。

   更多添加资源信息，请参见添加资源。命令示例如下：

   add py py_udtf_example.py;
   add file test_json.txt;
   add table table_resource1 as table_resource1;

4. 通过MaxCompute客户端，创建UDTF函数my_udtf。

   更多创建函数信息，请参见注册函数。命令示例如下：

   create function my_udtf as 'py_udtf_example.UDTFExample' using 'py_udtf_example.py, test_json.txt, table_resource1';

5. 通过MaxCompute客户端，基于新创建的UDTF函数，运行SQL语句。命令示例如下：
   • 示例1：单纯使用UDTF函数运行SQL。

     select my_udtf(pageid) as (pageid, adid) from tmp1;

     返回结果如下。

     +------------+------------+
     | pageid     | adid       |
     +------------+------------+
     | front_page | 1          |
     | front_page | 2          |
     | front_page | 3          |
     | contact_page1 | 3          |
     | contact_page1 | 4          |
     | contact_page1 | 5          |
     | contact_page3 | 5          |
     +------------+------------+

   • 示例2：对示例1中的命令改写，结合Lateral View运行SQL。

     select pageid, adid from tmp1 lateral view my_udtf(pageid) adTable as udtf_pageid, adid;

     返回结果如下。

     +------------+------------+
     | pageid     | adid       |
     +------------+------------+
     | front_page | 1          |
     | front_page | 2          |
     | front_page | 3          |
     | contact_page1 | 3          |
     | contact_page1 | 4          |
     | contact_page1 | 5          |
     | contact_page3 | 5          |
     +------------+------------+

     执行结果的两列分别为tmp1.pageid和UDTF输出的adid。

   • 示例3：结合聚合函数和Lateral View运行SQL。

     --聚合统计每个adid出现的次数。
     select adid, count(1) as cnt
         from tmp1 lateral view my_udtf(pageid) adTable as udtf_pageid, adid
     group by adid;

     返回结果如下。

     +------------+------------+
     | adid       | cnt        |
     +------------+------------+
     | 1          | 1          |
     | 2          | 1          |
     | 3          | 2          |
     | 4          | 1          |
     | 5          | 2          |
     +------------+------------+