MaxCompute:Develop a UDF in Python 2

Encoding declaration

Add an encoding declaration at the top of any UDF file that contains Chinese characters. Without it, MaxCompute returns an error at runtime. Both formats below are equivalent:

#coding:utf-8

# -*- coding: utf-8 -*-

Module import

Every UDF must import the function signature module:

from odps.udf import annotate

To reference files or tables in UDF code, also import from odps.distcache:

from odps.distcache import get_cache_file   # for file resources
from odps.distcache import get_cache_table  # for table resources

Function signature

The @annotate decorator defines the data types of input parameters and return value. MaxCompute checks type consistency during semantic parsing and returns an error if types do not match.

@annotate("bigint,bigint->bigint")

For the full signature syntax, see Function signatures and data types.

Minimal example

The following example shows a complete, working UDF that adds two integers. It covers all required components.

#coding:utf-8
# Import the function signature.
from odps.udf import annotate

# Define input types (BIGINT, BIGINT) and return type (BIGINT).
@annotate("bigint,bigint->bigint")
class MyPlus(object):
    def evaluate(self, arg0, arg1):
        if None in (arg0, arg1):
            return None
        return arg0 + arg1

Prohibited operations

MaxCompute runs Python 2 UDF code inside a sandbox. The following operations are not permitted:

• Reading from or writing to local files

• Starting subprocesses

• Starting threads

• Opening socket connections

• Calling Python 2 UDFs from external systems

Because of these restrictions, all uploaded code must rely on Python standard libraries. Modules or C extension modules that perform the prohibited operations above cannot be used.

Available standard library modules

All pure-Python modules in the Python standard library (those with no dependency on C extension modules) are available.

The following C extension modules are also available:

• array, audioop

• binascii, bisect

• cmath, _codecs_cn, _codecs_hk, _codecs_iso2022, _codecs_jp, _codecs_kr, _codecs_tw, _collections, cStringIO

• datetime

• _functools, future_builtins

• _heapq, _hashlib

• itertools

• _json

• _locale, _lsprof

• math, _md5, _multibytecodec

• operator

• _random

• _sha256, _sha512, _sha, _struct, strop

• time

• unicodedata

• _weakref

• cPickle

Signature format

@annotate('<arg_type_list>-><return_type>')

arg_type_list specifies input parameter types, separated by commas. It accepts two special forms:

• * — accepts any number of input parameters

• '' (empty string) — accepts no input parameters

return_type specifies the type of the single return value.

Supported input and return types: BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(precision,scale), CHAR, VARCHAR, and the complex types ARRAY, MAP, STRUCT (including nested complex types).

Signature examples

Data type mappings

Write Python UDF logic using the Python 2 types that correspond to MaxCompute SQL types. Type mismatches cause runtime errors.

Additional notes:

• NULL in MaxCompute SQL maps to None in Python 2.

• The silent parameter is added to odps.udf.int(value). If silent is set to True and the value cannot be converted to int, the function returns None instead of raising an error.

Reference a file

get_cache_file(resource_name) returns a file-like object with the content of the specified file resource.

• resource_name must be the name of an existing file resource in your MaxCompute project. If the name is invalid or the file does not exist, an error is returned.

• Declare the file resource when you create the UDF. If you do not declare it, calling the UDF returns an error.

• Call close() on the returned object when you are done with it.

from odps.udf import annotate
from odps.distcache import get_cache_file

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

Reference a table

get_cache_table(resource_name) returns a generator. Each iteration yields one record as a list.

• resource_name must be the name of an existing table resource in your MaxCompute project. If the name is invalid or the table does not exist, an error is returned.

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)