MaxCompute:Python 3 UDTF

MaxCompute Studioを使用して、Python 3でUDTFコードを記述できます。 コードには次の情報を含めることができます。

• モジュールのインポート: 必須。

  UDTFコードには、from odps.udf import annotateとfrom odps.udf import BaseUDTFを含める必要があります。 from odps.udf import annotateは、関数署名モジュールをインポートするために使用されます。 これにより、MaxComputeはコードで定義されている関数シグネチャを識別できます。 from odps.udf import BaseUDTFは、Python UDTFの基本クラスです。 このクラスを使用して、派生クラスにprocessやcloseなどのメソッドを実装する必要があります。

  UDTFコードでファイルまたはテーブルを参照する場合は、UDTFコードにfrom odps.distcache import get_cache_fileまたはfrom odps.distcache import get_cache_tableを含める必要があります。

• 関数署名: オプション。

  関数シグネチャは、@ annotate(<signature>) 形式です。 signatureパラメーターは、UDTFの入力パラメーターと戻り値のデータ型を定義するために使用されます。 関数シグネチャを指定しない場合、SQL文でUDTFを呼び出すときに、任意のデータ型の入力パラメーターを一致させることができます。 その結果、戻り値のデータ型は推論できず、すべての出力パラメーターはSTRING型になります。 関数シグネチャの詳細については、「関数シグネチャとデータ型」をご参照ください。

• カスタムPythonクラス (派生クラス): 必須です。

  カスタムPythonクラスは、UDTFコードの組織単位です。 このクラスは、ビジネス要件を満たすために使用される変数とメソッドを定義します。 UDTFコードでは、MaxComputeに構築されているサードパーティライブラリを参照したり、ファイルやテーブルを参照したりできます。 詳細については、「サードパーティライブラリ」または「リファレンスリソース」をご参照ください。

• Pythonクラスを実装するメソッド: required.

  Pythonクラスを実装するには、4つのメソッドを使用できます。 これらのメソッドを次の表に示します。

  移動方法	説明
  BaseUDTF.init()	初期化メソッド。 このメソッドを派生クラスに実装するには、コードの実行を開始するときに、基本クラスのsuper(BaseUDTF, self).init() 初期化メソッドを呼び出す必要があります。 INITメソッドは、UDTFのライフサイクル全体で1回だけ呼び出されます。 このメソッドは、最初のレコードが処理される前にのみ呼び出されます。 UDTFが内部状態を保存する必要がある場合、この方法を使用してすべての状態を初期化できます。
  BaseUDTF.process([args, ...])	process関数は、SQLレコードごとに1回呼び出されます。 process関数のパラメーターは、SQL文で指定されたUDTFの入力パラメーターです。
  BaseUDTF.forward([args, ...])	UDTFの出力メソッド。 このメソッドはユーザーコードによって呼び出されます。 forwardメソッドが呼び出されるたびに1つの出力レコードが生成されます。 forwardメソッドのパラメーターは、SQL文で指定されたUDTF出力パラメーターです。 Pythonコードで関数シグネチャが指定されていない場合、forwardメソッドが呼び出されるときにすべての出力値をSTRING型に変換する必要があります。
  BaseUDTF.close()	UDTFを終了するメソッド。 このメソッドは一度だけ呼び出されます。 最後のレコードが処理される前にのみ呼び出されます。

次の例は、UDTFコードを示しています。

# Import the function signature module and the base class. 
from odps.udf import annotate
from odps.udf import BaseUDTF
# The function signature. 
@annotate('string -> string')
# The custom Python class. 
class Explode(BaseUDTF):
# Methods used to implement the custom Python class. 
   def process(self, arg):
       props = arg.split(',')
       for p in props:
           self.forward(p)

Python Software Foundationは、Python 2のEOLを発表しました。 したがって、Python 2 UDTFを移植することを推奨します。 Python 2 UDTFを移植するために使用される方法は、MaxComputeプロジェクトのタイプによって異なります。

• 新しいプロジェクト: プロジェクトが新しいMaxComputeプロジェクトである場合、またはPythonを使用してプロジェクトに初めてUDTFを記述する場合は、Python 3を使用してすべてのPython UDTFを記述することを推奨します。

• 既存のプロジェクト: プロジェクトが多数のPython 2 UDTFが作成されている既存のプロジェクトである場合、Python 3を有効にするときは注意してください。 Python 2 UDTFをPython 3 UDTFに徐々に置き換える場合は、次のいずれかの方法を使用します。

  • Python 3を使用して新しいUDTFを作成し、セッションレベルで新しいジョブに対してPython 3を有効にします。 Python 3を有効にする方法の詳細については、「Python 3を有効にする」をご参照ください。

  • Python 2 UDTFを書き換えて、Python 2とPython 3の両方と互換性を持たせます。 UDTFの書き換え方法の詳細については、「Python 2コードのPython 3への移植」をご参照ください。

    説明

    パブリックUDTFを作成し、複数のMaxComputeプロジェクトにUDTFに対する権限を付与する必要がある場合は、UDTFがPython 2とPython 3の両方と互換性があることを確認することをお勧めします。

関数シグネチャの形式:

@annotate(<signature>)

signatureは関数署名文字列です。 このパラメーターは、入力パラメーターと戻り値のデータ型を識別するために使用されます。 UDTFを実行する場合、UDTFの入力パラメーターと戻り値は、関数シグネチャで指定されたものと同じデータ型である必要があります。 データ型の整合性は、セマンティック解析中にチェックされます。 データ型に矛盾がある場合は、エラーが返されます。 署名の形式は次のとおりです。

'arg_type_list -> type_list'

パラメーターの説明：

• type_list: 戻り値のデータ型を示します。 UDTFは複数の列を返すことができます。 BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision、scale) のデータ型がサポートされています。 ARRAY、MAP、STRUCTなどの複雑なデータ型、およびネストされた複雑なデータ型もサポートされています。

• arg_type_list: 入力パラメーターのデータ型を指定します。 複数の入力パラメーターを使用する場合は、複数のデータ型を指定し、コンマ (,) で区切ります。 BIGINT、STRING、DOUBLE、BOOLEAN、DATETIME、DECIMAL、FLOAT、BINARY、DATE、DECIMAL(precision、scale) 、CHAR、VARCHAR、複合データ型 (ARRAY、MAP、STRUCT) 、およびネストされた複合データ型がサポートされています。

  arg_type_listは、アスタリスク (*) または空のまま ('') で表すことができます。

  • arg_type_listがアスタリスク (*) で表される場合、ランダムな数の入力パラメーターが許可されます。

  • arg_type_listが空 ('') の場合、入力パラメーターは使用されません。

  @ Resolveアノテーションの構文拡張機能の詳細については、「UDAFおよびUDTFの動的パラメーター」をご参照ください。

次の表に、有効な関数シグネチャの例を示します。

次のサンプルコードは、ファイルとテーブルを参照する方法を示しています。

from odps.udf import annotate
from odps.udf import BaseUDTF
from odps.distcache import get_cache_file
from odps.distcache import get_cache_table
@annotate('string -> string, bigint')
class UDTFExample(BaseUDTF):
    """Read pageid and adid_list from the file and table to generate dict.
    """
    def __init__(self):
        import json
        cache_file = get_cache_file('test_json.txt')
        self.my_dict = json.load(cache_file)
        cache_file.close()
        records = list(get_cache_table('table_resource1'))
        for record in records:
            self.my_dict[record[0]] = record[1]
    """Enter pageid and generate pageid and all adid values.
    """
    def process(self, pageid):
        for adid in self.my_dict[pageid]:
            self.forward(pageid, adid)

概要 の手順に従ってPython 3 UDTFを開発した後、MaxCompute SQLを使用してPython 3 UDTFを呼び出すことができます。 次のいずれかのメソッドを使用して、Python 3 UDTFを呼び出すことができます。

• MaxComputeプロジェクトでUDFを使用する: この方法は、組み込み関数を使用する方法と似ています。

• プロジェクト間でUDFを使用する: プロジェクトaでプロジェクトBのUDFを使用します。次のステートメントは、例を示します。select B:udf_in_other_project(arg0, arg1) as res from table_t; プロジェクト間共有の詳細については、「パッケージに基づくプロジェクト間リソースアクセス」をご参照ください。