Realtime Compute for Apache Flink:Python ユーザー定義スカラー関数 (UDSF)

制限事項

Realtime Compute for Apache Flink で Python ユーザー定義関数 (UDF) を開発する場合、次の制約があります。

UDSF の作成

次の手順では、例として Windows 環境を使用します。Flink は、UDSF、ユーザー定義集計関数 (UDAF)、ユーザー定義テーブル値関数 (UDTF) の実装を含むサンプルリポジトリを提供しています。

1. python_demo-master をローカルマシンにダウンロードして解凍します。

   これはサードパーティの GitHub リポジトリです。アクセスが遅い、または断続的に利用できない場合があります。

2. PyCharm で、[ファイル] > [開く] を選択し、解凍した python_demo-master ディレクトリを開きます。

3. \python_demo-master\udx パスにある udfs.py を開き、UDSF を定義します。

   from pyflink.table import DataTypes
   from pyflink.table.udf import udf
   
   @udf(result_type=DataTypes.STRING())
   def sub_string(s: str, begin: int, end: int):
       return s[begin:end]

   sub_string の例では、入力文字列内の位置 begin から位置 end までの文字を抽出します。

4. \python_demo-master ディレクトリで、次のコマンドを実行して udx ディレクトリをパッケージ化します：

   zip -r python_demo.zip udx

   \python_demo-master\ に python_demo.zip が生成されたら、パッケージの準備は完了です。

UDSF の登録

パッケージを作成した後、Realtime Compute for Apache Flink コンソールで UDSF を登録します。登録手順については、「ユーザー定義関数 (UDF) の管理」をご参照ください。

UDSF の使用

UDSF を登録した後、Flink SQL ジョブで使用します。

1. Flink SQL を使用してドラフトを作成します。詳細については、「ジョブ開発の概要」をご参照ください。次の例では、ASI_UDSF (UDSF の登録名) を呼び出して、ソーステーブルの a フィールドの位置 2 から 4 の直前までの文字を抽出します：

   CREATE TEMPORARY TABLE ASI_UDSF_Source (
     a VARCHAR,
     b INT,
     c INT
   ) WITH (
     'connector' = 'datagen'
   );
   
   CREATE TEMPORARY TABLE ASI_UDSF_Sink (
     a VARCHAR
   ) WITH (
     'connector' = 'blackhole'
   );
   
   INSERT INTO ASI_UDSF_Sink
   SELECT ASI_UDSF(a, 2, 4)
   FROM ASI_UDSF_Source;

2. 開発コンソールの左側のナビゲーションペインで、[O&M] > [デプロイ] を選択します。対象のデプロイを見つけ、[アクション] 列で [開始] をクリックします。デプロイが開始されると、ASI_UDSF_Source の a フィールドの 2–4 番目の文字が ASI_UDSF_Sink に書き込まれます。

非同期ユーザー定義関数

外部データベースへのアクセスや HTTP リクエストなど、I/O 集中型の処理を実行する UDF には、非同期ユーザー定義関数を使用します。単一の非同期 UDF は複数の I/O リクエストを同時に処理できるため、待機時間をリクエスト間で分散し、ジョブのスループットを向上させます。

import asyncio

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


# 方法1：Python の async 関数を使用します
@udf(result_type=DataTypes.STRING())
async def async_api_call(product_id: str) -> str:
    await asyncio.sleep(0.05)
    return f"product_{product_id}"


# 方法2：非同期関数クラスをサブクラス化します
class AsyncUserLookup(AsyncScalarFunction):
    def open(self, function_context):
        self.cache = {}

    async def eval(self, user_id: str) -> str:
        if user_id in self.cache:
            return self.cache[user_id]

        await asyncio.sleep(0.05)
        result = f"user_{user_id}"
        self.cache[user_id] = result
        return result

    def close(self):
        self.cache.clear()


async_user_lookup = udf(
    AsyncUserLookup(),
    input_types=[DataTypes.STRING()],
    result_type=DataTypes.STRING()
)