Python - Realtime Compute for Apache Flink - Alibaba Cloud ドキュメントセンター

ユーザー定義関数のタイプ

カテゴリ	説明
ユーザー定義スカラー関数 (UDSF)	UDSF は、0 個、1 個、または複数のスカラー値を新しいスカラー値にマッピングします。1 つの入力行を処理して 1 つの出力値を生成し、1 対 1 のマッピングを作成します。詳細については、「ユーザー定義スカラー関数 (UDSF)」をご参照ください。
ユーザー定義集計関数 (UDAF)	UDAF は、複数のレコードを 1 つのレコードに集約し、多対 1 のマッピングを作成します。詳細については、「ユーザー定義集計関数 (UDAF)」をご参照ください。
ユーザー定義テーブル関数 (UDTF)	UDTF は、0 個、1 個、または複数のスカラー値を入力パラメーターとして受け取ります。スカラー関数とは異なり、テーブル関数は単一の値だけでなく、任意の数の行を出力として返すことができます。返される行は、1 つ以上の列で構成できます。単一の関数呼び出しで、複数の行または列を生成できます。詳細については、「ユーザー定義テーブル関数 (UDTF)」をご参照ください。

Python 依存関係の使用

Realtime Compute for Apache Flink のクラスターには、Pandas、NumPy、PyArrow などの一般的な Python パッケージがプリインストールされています。プリインストールされているサードパーティの Python パッケージのリストについては、「Python ジョブの開発」ページをご参照ください。関数で使用する前に、これらのパッケージをインポートする必要があります。次の例は、パッケージをインポートする方法を示しています。

@udf(result_type=DataTypes.FLOAT())
def percentile(values: List[float], percentile: float):
    import numpy as np
    return np.percentile(values, percentile)

プリインストールされていないサードパーティの Python パッケージを使用するには、Python UDF の登録時に依存関係ファイルとしてアップロードします。詳細については、「ユーザー定義関数 (UDF) の管理」および「Python 依存関係の使用」をご参照ください。

コードのデバッグ

logging モジュールを使用して、Python ユーザー定義関数からログ情報を出力できます。これにより、問題の特定と診断に役立ちます。次の例は、logging の使用方法を示しています。

@udf(result_type=DataTypes.BIGINT())
def add(i, j):    
  logging.info("hello world")    
  return i + j

ログが生成された後、TaskManager のログファイルで確認できます。詳細については、「実行ログの表示」をご参照ください。

パフォーマンスチューニング

リソースのプリロード

関数の初期化中にリソースをプリロードすることで、eval メソッドが呼び出されるたびにリソースが再ロードされるのを回避できます。たとえば、大規模な深層学習モデルを一度だけロードし、そのモデルで複数回バッチ予測を実行できます。

from pyflink.table import DataTypes
from pyflink.table.udf import ScalarFunction, udf

class Predict(ScalarFunction):
    def open(self, function_context):
        import pickle

        with open("resources.zip/resources/model.pkl", "rb") as f:
            self.model = pickle.load(f)

    def eval(self, x):
        return self.model.predict(x)

predict = udf(Predict(), result_type=DataTypes.DOUBLE(), func_type="pandas")

説明

Python データファイルのアップロード方法については、「Python 依存関係の使用」をご参照ください。

非同期ユーザー定義関数

外部データベースへのアクセスや HTTP サービスコールなど、I/O 集約型のシナリオでは、非同期ユーザー定義関数を使用します。単一の関数インスタンスが複数のリクエストを同時に処理できるため、待機時間が複数の呼び出しに分散され、ジョブのスループットが大幅に向上します。この機能は VVR 11.7 以降のバージョンでのみサポートされており、ユーザー定義スカラー関数 (UDSF) にのみ適用されます。詳細については、「非同期ユーザー定義関数」をご参照ください。

Pandas ライブラリの使用

Realtime Compute for Apache Flink は、標準の Python ユーザー定義関数に加えて、Pandas ユーザー定義関数もサポートしています。これらの関数の入力データ型は、pandas.Series や pandas.DataFrame などの Pandas データ構造です。これらの関数内で Pandas や NumPy などの高性能な Python ライブラリを使用して、効率的な Python ユーザー定義関数を開発できます。詳細については、「ベクター化ユーザー定義関数」をご参照ください。

パラメーター

Python ユーザー定義関数のパフォーマンスは、その実装に大きく依存します。パフォーマンスの問題が発生した場合は、関数の実装を最適化してください。次のパラメーターもパフォーマンスに影響します。

パラメーター	説明
python.fn-execution.bundle.size	Python UDF は非同期で実行されます。Java オペレーターは、実行のために Python プロセスにデータを送信する前にデータをキャッシュします。キャッシュが特定のしきい値に達すると、データが Python プロセスに送信されます。`python.fn-execution.bundle.size` パラメーターは、キャッシュできるデータレコードの最大数を制御します。デフォルト値は 100,000 レコードです。
python.fn-execution.bundle.time	このパラメーターは、データの最大キャッシュ時間を制御します。キャッシュされたデータは、レコード数が `python.fn-execution.bundle.size` のしきい値に達するか、キャッシュ時間が `python.fn-execution.bundle.time` のしきい値に達したときに処理のために送信されます。デフォルト値は 1,000 ミリ秒です。
python.fn-execution.arrow.batch.size	Pandas UDF の場合、このパラメーターは Arrow バッチに含めることができるレコードの最大数を指定します。デフォルト値は 10,000 です。説明 `python.fn-execution.arrow.batch.size` パラメーターの値は、`python.fn-execution.bundle.size` パラメーターの値より大きくすることはできません。

説明

これらのパラメーターを過度に大きな値に設定すると、逆効果になる可能性があります。チェックポイント中にバッファリングされるデータが多すぎると、チェックポイントが長くなったり、失敗したりする可能性があります。これらのパラメーターの詳細については、「設定」をご参照ください。

Realtime Compute for Apache Flink:Python