Realtime Compute for Apache Flink:Python

UDF の分類

Python 依存関係の使用

pandas、NumPy、PyArrow など、一般的に使用される Python パッケージは、Realtime Compute for Apache Flink のワークスペースにプリインストールされています。 Realtime Compute for Apache Flink のワークスペースにプリインストールされているサードパーティ製 Python パッケージの詳細については、Python API ドラフトの開発 をご参照ください。プリインストールされている Python パッケージは、Python UDF にインポートする必要があります。サンプルコード：

@udf(result_type=DataTypes.FLOAT())
def percentile(values: List[float], percentile: float):
    # numpy をインポートします
    import numpy as np
    return np.percentile(values, percentile)

Python UDF では、他の種類のサードパーティ製 Python パッケージも使用できます。フルマネージド Flink にプリインストールされていないサードパーティ製 Python パッケージを使用する場合は、そのパッケージを依存関係ファイルとしてフルマネージド Flink にアップロードする必要があります。詳細については、UDF の管理 および Python 依存関係の使用 をご参照ください。

コードのデバッグ

Python UDF のコード実装では、ログ生成メソッドを使用してログを生成し、ログに基づいてエラーを特定できます。次のコードは例です。

@udf(result_type=DataTypes.BIGINT())
def add(i, j):    
  # ログを生成します
  logging.info("hello world")    
  return i + j

ログが生成された後、TaskManager のログファイルでログを表示できます。詳細については、デプロイメントの操作ログの表示 をご参照ください。

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

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

# 予測クラスを定義します
class Predict(ScalarFunction):
    def open(self, function_context):
        # pickle をインポートします
        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")

参照

• UDF の登録、更新、削除方法の詳細については、UDF の管理 をご参照ください。

• Python UDF の開発と使用方法の詳細については、UDAF、UDSF、および UDTF をご参照ください。

• Flink Python デプロイメントでカスタム Python 仮想環境、サードパーティ製 Python パッケージ、JAR パッケージ、およびデータファイルを使用する方法の詳細については、Python 依存関係の使用 をご参照ください。

• Java UDF の開発と使用方法の詳細については、UDAF、UDSF、および UDTF をご参照ください。

• Java UDF のデバッグとチューニング方法の詳細については、Java をご参照ください。

• UDAF を使用してデータをソートおよび集計する方法の詳細については、UDAF を使用してデータをソートおよび集計する をご参照ください。