MaxCompute:UDF とサードパーティ製 Python ライブラリの使用

DataFrame では、Sequence オブジェクトに対して map メソッドを使用して、そのすべての要素に対して UDF を呼び出すことができます。

>>> iris.sepallength.map(lambda x: x + 1).head(5)
   sepallength
0          6.1
1          5.9
2          5.7
3          5.6
4          6.0

map メソッドの実行後に Sequence の型が変更された場合は、Sequence の新しい型を明示的に指定する必要があります。

>>> iris.sepallength.map(lambda x: 't'+str(x), 'string').head(5)
   sepallength
0         t5.1
1         t4.9
2         t4.7
3         t4.6
4         t5.0

関数がクロージャを含む場合、関数外のクロージャ変数の値の変更によって、関数内の変数の値が変更されます。

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

その結果、dfs 内の各 SequenceExpr オブジェクトは df.sepal_length + 9 になります。この問題を解決するには、関数を別の関数の戻り値として使用するか、partial を使用します。次の 2 つの例を参照してください。

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

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

カウンターの使用例：

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)

LogView の JSONSummary でカウンターの値を確認できます。

1 行に対して UDF を使用するには、apply メソッドを使用します。axis パラメーターを 1 に設定して、操作が行に対して実行されることを示す必要があります。apply メソッドの UDF は、先行する Collection オブジェクトからのデータ行であるパラメーターを受け取ります。属性またはオフセットを使用して、フィールド内のデータを取得できます。

• reduce パラメーターが True に設定されている場合、Sequence オブジェクトが返されます。それ以外の場合、Collection オブジェクトが返されます。names パラメーターと types パラメーターを使用して、返される Sequence オブジェクトまたは Collection オブジェクトのフィールド名と型を指定できます。デフォルトでは、型を指定しない場合、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

• reduce が apply メソッドの UDF で False の場合、yield キーワードを使用して複数の行の結果を返すことができます。

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

• 返されるフィールド名と型を関数にアノテーションすることもできます。関数を呼び出すときに指定する必要はありません。

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

• map-only の map_reduce を使用することもできます。これは、apply メソッドで axis=1 を指定した場合と同じです。

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

• MaxCompute で既存のユーザー定義テーブル関数（UDTF）を使用するには、UDTF の名前を指定します。

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

• 行に対して apply メソッドを使用し、reduce が False の場合、既存の行で lateral view を使用して、集計などの目的で使用できます。

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

apply メソッドを使用する場合、axis が指定されていないか、axis の値が 0 の場合、カスタム集計クラスを渡してすべての Sequence オブジェクトを集計できます。

class Agg(object):

    def buffer(self):
        return [0.0, 0]

    def __call__(self, buffer, val):
        buffer[0] += val
        buffer[1] += 1

    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
        buffer[1] += pbuffer[1]

    def getvalue(self, buffer):
        if buffer[1] == 0:
            return 0.0
        return buffer[0] / buffer[1]

>>> iris.exclude('name').apply(Agg)
   sepallength_aggregation  sepalwidth_aggregation  petallength_aggregation  petalwidth_aggregation
0                 5.843333                   3.054                 3.758667                1.198667

UDF は、MaxCompute リソース（テーブルリソースやファイルリソースなど）を読み取ったり、Collection オブジェクトをリソースとして参照したりすることもできます。これを行うには、UDF をクロージャまたは呼び出し可能なクラスとして記述する必要があります。次の 2 つの例を参照してください。

>>> 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 the values by using the 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

行操作で axis が 1 の場合、関数クロージャまたは呼び出し可能なクラスを記述する必要があります。列の集計操作の場合は、__init__ 関数を使用してリソースを読み取るだけで済みます。

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