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

クロージャ変数のキャプチャバグの回避

UDF にクロージャが含まれている場合、キャプチャされた変数への外部からの変更が関数の動作に影響します。以下のコードは意図しない結果になります。 dfs 内の各 SequenceExpr は最終的に df.sepal_length + 9 になります。

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

これを修正するには、外側の関数からラムダを返すか、functools.partial を使用してください。

# オプション 1：ファクトリ関数を使用
>>> dfs = []
>>> def get_mapper(i):
>>>     return lambda x: x + i
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(get_mapper(i)))

# オプション 2：functools.partial を使用
>>> import functools
>>> dfs = []
>>> for i in range(10):
>>>     dfs.append(df.sepal_length.map(functools.partial(lambda v, x: x + v, i)))

既存の UDF の使用

関数名 (文字列) または Function オブジェクトを map に渡すことで、既存の UDF を呼び出せます。詳細については、「関数」をご参照ください。

カウンタによる実行の追跡

get_execution_context を使用して、UDF 内からカウンタにアクセスします。カウンタの値は LogView の JSONSummary に表示されます。

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)

行ごとに単一の値を返す

reduce=True を設定すると、Sequence が返ります。出力型は types パラメータで指定します (デフォルトは 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

yield を使用した複数行の返却

reduce=False を設定し、yield を使用して入力行ごとに複数の行を出力します。出力フィールド名と型は names と 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_sub', 'iris_add'], types=['float', 'float']).count()
300

呼び出し側で繰り返し指定するのを避けるため、関数に直接出力スキーマを注釈として付与できます。

>>> from odps.df import output

>>> @output(['iris_sub', 'iris_add'], ['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 のみの map_reduce

map のみのモードの map_reduce は、apply と axis=1 と同等です。

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

既存の UDTF の使用

MaxCompute の既存のユーザー定義テーブル値関数 (UDTF) を呼び出すには、その名前を文字列として渡します。

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

ラテラルビューによる行出力と元の列の結合

reduce=False の場合、ラテラルビューを使用して UDF 出力を元の列と結合できます。これは集計に便利です。

>>> from odps.df import output

>>> @output(['iris_sub', 'iris_add'], ['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)]

ファイルリソースと Collection リソースを使用した行レベル UDF

>>> 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):  # リソースは呼び出し時の指定順で渡されます
>>>     names = set()
>>>     fileobj = resources[0]  # ファイルリソースはファイルライクオブジェクトとして渡されます
>>>     for l in fileobj:
>>>         names.add(l)
>>>     collection = resources[1]
>>>     for r in collection:
>>>         names.add(r.name)  # フィールド名またはオフセットで値を取得します
>>>     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

パーティション分割テーブルを読み取る場合、パーティションフィールドは含まれません。

ローカル DataFrame をリソースとして使用した行レベル UDF

ローカル変数は、実行時に MaxCompute のリソースとして参照できます。次の例では、stop_words はローカル DataFrame であり、実行プログラムがそれをリソースとして UDF に渡します。

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

行操作 (axis=1) の場合、関数クロージャまたは呼び出し可能クラスを使用してリソースをロードしてください。列集計の場合は、代わりに __init__ メソッドを使用してください。

Pure Python パッケージ

デフォルトでは、PyODPS は Pure Python コードを含むがファイル操作を含まないサードパーティライブラリをサポートしています。次の例では、python-dateutil とその依存関係である six をアップロードします。

ステップ 1：パッケージとその依存関係をダウンロードします。パッケージは Linux 用にビルドされている必要があります。

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

これにより、six-1.10.0-py2.py3-none-any.whl と python_dateutil-2.5.3-py2.py3-none-any.whl がダウンロードされます。

ステップ 2：create_resource を使用して両方のファイルをリソースとしてアップロードします。

# ファイルの拡張子が正しいことを確認してください。
>>> 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'))

ステップ 3：ライブラリを使用します。options.df.libraries でセッション全体にグローバルに指定するか、libraries パラメータを実行メソッドに直接渡して適用範囲を単一の呼び出しに限定します。以下の例では、日付文字列を含むサンプル DataFrame を使用します。

# グローバル設定 (このセッション内の以降のすべての DataFrame 操作に適用されます)
>>> 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_dates.datestr.map(get_year)
   datestr
0     2016
1     2015

# 実行ごとの設定 (この呼び出しにのみ適用されます)
>>> def get_year(t):
>>>     from dateutil.parser import parse
>>>     return parse(t).strftime('%Y')

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

バイナリパッケージ (コンパイル済みコードを含む)

コンパイル済み C 拡張機能を含むパッケージ (SciPy や pandas など) には、追加の手順が必要です。

• .whl ファイルは cp27-cp27m-manylinux1_x86_64 プラットフォームタグを使用する必要があります。

• ファイルをアーカイブリソースとしてアップロードし、.whl 拡張子を .zip に変更します。

• odps.isolation.session.enable を True に設定するか、プロジェクト設定で isolation を有効にします。

# バイナリパッケージを .zip 拡張子のアーカイブとしてアップロードします。
>>> odps.create_resource('scipy.zip', 'archive', file_obj=open('scipy-0.19.0-cp27-cp27m-manylinux1_x86_64.whl', 'rb'))

# プロジェクトで既に分離が有効になっている場合、次のオプションは不要です。
>>> options.sql.settings = {'odps.isolation.session.enable': True}

>>> def psi(value):
>>>     # オペレーティングシステム間の構造の違いによるエラーを避けるため、
>>>     # サードパーティライブラリは関数内でインポートしてください。
>>>     from scipy.special import psi
>>>     return float(psi(value))

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

ソースからバイナリパッケージをビルドするには、Linux シェルで次のコマンドを実行してください。macOS または Windows でビルドされた Wheel ファイルは MaxCompute と互換性がありません。

python setup.py bdist_wheel

MaxCompute コンソールによるアップロード

PyODPS API の代わりに、MaxCompute コンソールで add archive を使用してパッケージをアップロードすることもできます。

ステップ 1：各依存関係の正しいパッケージファイルを特定します。

ほとんどのパッケージは、複数のプラットフォーム用の .whl ファイルを提供しています。バイナリパッケージの場合、名前に cp27-cp27m-manylinux1_x86_64 が含まれるファイルを見つけてください。Pure Python パッケージの場合、py2.py3-none-any の Wheel が使用できます。

ステップ 2：必要なすべての依存関係を確認します。次の表は、一般的なパッケージの依存関係を示しています。

NumPy は既にプリインストールされています。python-dateutil、pytz、pandas、SciPy、scikit-learn、six のみをアップロードしてください。

ステップ 3：パッケージファイルをダウンロードします。次の表は、各パッケージでダウンロードする特定のファイルを示しています。

ステップ 4：各ファイルをアップロードします。バイナリパッケージ (pandas、SciPy、scikit-learn) の場合、アップロード前に .whl 拡張子を .zip に変更してください。

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