MaxCompute:Use UDFs and the third-party Python libraries

Avoid closure variable capture bugs

When a UDF contains a closure, external changes to the captured variable affect the function's behavior. The following code produces an unintended result — each SequenceExpr in dfs ends up as df.sepal_length + 9:

>>> dfs = []
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(lambda x: x + i))

Fix this by returning the lambda from an outer function, or by using functools.partial:

# Option 1: use a factory function
>>> dfs = []
>>> def get_mapper(i):
>>>     return lambda x: x + i
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(get_mapper(i)))

# Option 2: use functools.partial
>>> import functools
>>> dfs = []
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(functools.partial(lambda v, x: x + v, i)))

Use existing UDFs

Pass a function name (string) or a Function object to map to call an existing UDF. For details, see Functions.

Track execution with counters

Use get_execution_context to access counters from within a UDF. Counter values appear in JSONSummary of LogView.

from odps.udf import get_execution_context

def h(x):
    ctx = get_execution_context()
    counters = ctx.get_counters()
    counters.get_counter('df', 'add_one').increment(1)
    return x + 1

df.field.map(h)

Return a single value per row

Set reduce=True to return a Sequence. Specify the output type with the types parameter (default is STRING).

>>> iris.apply(lambda row: row.sepallength + row.sepalwidth, axis=1, reduce=True, types='float').rename('sepaladd').head(3)
   sepaladd
0       8.6
1       7.9
2       7.9

Return multiple rows using yield

Set reduce=False and use yield to emit multiple rows per input row. Specify output field names and types with names and types.

>>> iris.count()
150

>>> def handle(row):
>>>     yield row.sepallength - row.sepalwidth, row.sepallength + row.sepalwidth
>>>     yield row.petallength - row.petalwidth, row.petallength + row.petalwidth

>>> iris.apply(handle, axis=1, names=['iris_add', 'iris_sub'], types=['float', 'float']).count()
300

Annotate the output schema directly on the function to avoid repeating it at call sites:

>>> from odps.df import output

>>> @output(['iris_add', 'iris_sub'], ['float', 'float'])
>>> def handle(row):
>>>     yield row.sepallength - row.sepalwidth, row.sepallength + row.sepalwidth
>>>     yield row.petallength - row.petalwidth, row.petallength + row.petalwidth

>>> iris.apply(handle, axis=1).count()
300

Equivalent: map-only map_reduce

map_reduce in map-only mode is equivalent to apply with axis=1:

>>> iris.map_reduce(mapper=handle).count()
300

Use an existing UDTF

To call an existing user-defined table-valued function (UDTF) in MaxCompute, pass its name as a string:

>>> iris['name', 'sepallength'].apply('your_func', axis=1, names=['name2', 'sepallength2'], types=['string', 'float'])

Combine row output with a lateral view

When reduce=False, combine the UDF output with original columns using a lateral view — useful for aggregations:

>>> from odps.df import output

>>> @output(['iris_add', 'iris_sub'], ['float', 'float'])
>>> def handle(row):
>>>     yield row.sepallength - row.sepalwidth, row.sepallength + row.sepalwidth
>>>     yield row.petallength - row.petalwidth, row.petallength + row.petalwidth

>>> iris[iris.category, iris.apply(handle, axis=1)]

Row-level UDF with file and collection resources

>>> file_resource = o.create_resource('pyodps_iris_file', 'file', file_obj='Iris-setosa')

>>> iris_names_collection = iris.distinct('name')[:2]
>>> iris_names_collection
       sepallength
0      Iris-setosa
1  Iris-versicolor

>>> def myfunc(resources):  # resources are passed in by calling order
>>>     names = set()
>>>     fileobj = resources[0]  # file resources are represented by a file-like object
>>>     for l in fileobj:
>>>         names.add(l)
>>>     collection = resources[1]
>>>     for r in collection:
>>>         names.add(r.name)  # retrieve values by field name or offset
>>>     def h(x):
>>>         if x in names:
>>>             return True
>>>         else:
>>>             return False
>>>     return h

>>> df = iris.distinct('name')
>>> df = df[df.name,
>>>         df.name.map(myfunc, resources=[file_resource, iris_names_collection], rtype='boolean').rename('isin')]

>>> df
              name   isin
0      Iris-setosa   True
1  Iris-versicolor   True
2   Iris-virginica  False

When reading partitioned tables, partition fields are not included.

Row-level UDF with a local DataFrame as a resource

Local variables can be referenced as resources in MaxCompute at execution time. In the following example, stop_words is a local DataFrame that the executor passes to the UDF as a resource:

>>> words_df
                     sentence
0                 Hello World
1                Hello Python
2  Life is short I use Python

>>> import pandas as pd
>>> stop_words = DataFrame(pd.DataFrame({'stops': ['is', 'a', 'I']}))

>>> @output(['sentence'], ['string'])
>>> def filter_stops(resources):
>>>     stop_words = set([r[0] for r in resources[0]])
>>>     def h(row):
>>>         return ' '.join(w for w in row[0].split() if w not in stop_words),
>>>     return h

>>> words_df.apply(filter_stops, axis=1, resources=[stop_words])
                sentence
0            Hello World
1           Hello Python
2  Life short use Python

For row operations (axis=1), use a function closure or callable class to load resources. For column aggregations, use the __init__ method instead.

Pure Python packages

By default, PyODPS supports third-party libraries that contain pure Python code but no file operations. The following example uploads python-dateutil and its dependency six.

Step 1: Download the package and its dependencies. Packages must be built for Linux.

$ pip download python-dateutil -d /to/path/

This downloads six-1.10.0-py2.py3-none-any.whl and python_dateutil-2.5.3-py2.py3-none-any.whl.

Step 2: Upload both files as resources using create_resource.

# Make sure that file name extensions are correct.
>>> odps.create_resource('six.whl', 'file', file_obj=open('six-1.10.0-py2.py3-none-any.whl', 'rb'))
>>> odps.create_resource('python_dateutil.whl', 'file', file_obj=open('python_dateutil-2.5.3-py2.py3-none-any.whl', 'rb'))

Step 3: Use the libraries. Specify them globally via options.df.libraries, or per-execution via the libraries parameter.

# Global configuration (applies to all subsequent DataFrame operations in this session)
>>> from odps import options

>>> def get_year(t):
>>>     from dateutil.parser import parse
>>>     return parse(t).strftime('%Y')

>>> options.df.libraries = ['six.whl', 'python_dateutil.whl']
>>> df.datestr.map(get_year)
   datestr
0     2016
1     2015

# Per-execution configuration (applies only to this call)
>>> def get_year(t):
>>>     from dateutil.parser import parse
>>>     return parse(t).strftime('%Y')

>>> df.datestr.map(get_year).execute(libraries=['six.whl', 'python_dateutil.whl'])
   datestr
0     2016
1     2015

Binary packages (containing compiled code)

Packages that include compiled C extensions (such as SciPy or pandas) require additional steps:

• The .whl file must use the cp27-cp27m-manylinux1_x86_64 platform tag.

• Upload the file as an archive resource, with the .whl extension renamed to .zip.

• Set odps.isolation.session.enable to True, or enable isolation in your project settings.

# Upload the binary package as an archive with the .zip extension.
>>> odps.create_resource('scipy.zip', 'archive', file_obj=open('scipy-0.19.0-cp27-cp27m-manylinux1_x86_64.whl', 'rb'))

# If isolation is already enabled in your project, the following option is optional.
>>> options.sql.settings = {'odps.isolation.session.enable': True}

>>> def psi(value):
>>>     # Import the third-party library inside the function to avoid errors
>>>     # caused by structural differences between operating systems.
>>>     from scipy.special import psi
>>>     return float(psi(value))

>>> df.float_col.map(psi).execute(libraries=['scipy.zip'])

To build a binary package from source, run the following in a Linux shell. Wheel files built on macOS or Windows are not compatible with MaxCompute.

python setup.py bdist_wheel

Upload via the MaxCompute console

As an alternative to the PyODPS API, upload packages using add archive in the MaxCompute console.

Step 1: Identify the correct package file for each dependency.

Most packages provide .whl files for multiple platforms. For binary packages, find the file with cp27-cp27m-manylinux1_x86_64 in its name. For pure Python packages, any py2.py3-none-any wheel works.

Step 2: Verify all required dependencies. The following table lists dependencies for common packages.

NumPy is already pre-installed. Upload only python-dateutil, pytz, pandas, SciPy, scikit-learn, and six.

Step 3: Download the package files. The following table lists the specific files to download for each package.

Step 4: Upload each file. For binary packages (pandas, SciPy, scikit-learn), rename the .whl extension to .zip before uploading.

add archive python-dateutil.zip;
add archive pandas.zip;