MaxCompute:ユースケース

Python UDF スクリプトを作成し、sum_pandas.py ファイルとして保存します。以下のコードはその例です。

from odps.udf import annotate
import pandas as pd

@annotate("string, string -> string")
class SumColumns(object):
    def evaluate(self, arg1, arg2):
        # 入力引数を pandas DataFrame に変換します。
        df = pd.DataFrame({'col1': arg1.split(','), 'col2': arg2.split(',')})

        # pandas を使用してデータを処理します。
        # この例では、2 つのカラムの合計を計算します。
        df['sum'] = df['col1'].astype(int) + df['col2'].astype(int)

        # 結果を文字列に変換して返します。
        result = ','.join(df['sum'].astype(str).values)
        return result

sum_pandas.py スクリプトをリソースとしてご利用の MaxCompute プロジェクトにアップロードします。詳細については、「リソースの追加」をご参照ください。スクリプトをアップロードするサンプルコマンドは以下のとおりです。

ADD PY sum_pandas.py -f;

アップロードした sum_pandas.py スクリプトを SumColumns ユーザー定義関数として登録します。詳細については、「関数の登録」をご参照ください。関数を登録するサンプルコマンドは以下のとおりです。

CREATE FUNCTION SumColumns AS 'sum_pandas.SumColumns' USING 'sum_pandas.py';

testsum という名前のテストテーブルを作成し、テストデータを挿入します。

CREATE TABLE testsum (col1 string, col2 string);
INSERT INTO testsum VALUES ('1,2,3','1,2,3'),('1,2,3','3,2,1'),('1,2,3','4,5,6');

UDF を呼び出し、フラグを使用して既存のイメージを指定します。

set odps.sql.python.version=cp37;
set odps.session.image = <image_name>;
SELECT SumColumns(col1,col2) AS result FROM testsum;

以下の結果が返されます。

+------------+
| result     |
+------------+
| 2,4,6      |
| 4,4,4      |
| 5,7,9      |
+------------+

test_float_col という名前のテストテーブルを作成し、テストデータを挿入します。

CREATE TABLE test_float_col (col1 double);
INSERT INTO test_float_col VALUES (3.75),(2.51);

psi(col1) の値を計算する PyODPS スクリプトを作成し、psi_col.py ファイルとして保存して実行します。以下のコードはその例です。

import os
from odps import ODPS, options

def my_psi(v):
    from scipy.special import psi

    return float(psi(v))

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

o = ODPS(
      # ALIBABA_CLOUD_ACCESS_KEY_ID 環境変数にご利用の AccessKey ID を、
      # ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数にご利用の AccessKey Secret を設定してください。
      # AccessKey ID および AccessKey Secret をコード内にハードコードしないことを推奨します。
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
      project='your-default-project',
      endpoint='your-end-point'
)

df = o.get_table("test_float_col").to_df()
# ジョブを実行して結果を取得します。
df.col1.map(my_psi).execute(image='scipy')
# 結果を別のテーブルに保存します。
df.col1.map(my_psi).persist("result_table", image='scipy')

result_table テーブル内の結果を確認します。

SELECT * FROM result_table

以下の結果が返されます。

+------------+
| col1       |
+------------+
| 1.1825373886117962 |
| 0.7080484451910534 |
+------------+

test_float_col という名前のテストテーブルを作成し、テストデータを挿入します。

CREATE TABLE test_float_col (col1 double);
INSERT INTO test_float_col VALUES (3.75),(2.51);

psi(col1) の値を計算する MaxFrame スクリプトを作成し、psi_col.py ファイルとして保存して実行します。以下のコードはその例です。

import os
from odps import ODPS, options
from maxframe.session import new_session
import maxframe.dataframe as md


from maxframe.config import options
from maxframe import config

# 組み込みの SciPy イメージを参照します。
config.options.sql.settings = {
    "odps.session.image": "scipy"
}
def my_psi(v):
    from scipy.special import psi
    return float(psi(v))

o = ODPS(
      # ALIBABA_CLOUD_ACCESS_KEY_ID 環境変数にご利用の AccessKey ID を、
      # ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数にご利用の AccessKey Secret を設定してください。
      # AccessKey ID および AccessKey Secret をコード内にハードコードしないことを推奨します。
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
      os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
      project='your-default-project',
      endpoint='your-end-point'
)

# MaxFrame セッションを作成します。
session = new_session(o)
df = md.read_odps_table('test_float_col')

# ジョブを実行して結果を取得します。
print(df.col1.map(my_psi).execute().fetch())

以下の結果が返されます。

0    1.182537
1    0.708048
Name: col1, dtype: float64