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

制限事項

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

開発方法

Python UDSF は、次のいずれかの方法で開発できます。

• Python コードをパッケージ化し、パッケージをアップロードして、プラットフォームに関数として登録します。

• SQL 文でコードロジックをインライン関数として宣言します。

UDSF のロジックが単純な場合は、インライン関数として開発することを検討してください。

UDSFの作成

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

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

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

2. PyCharm で、[File] > [Open] を選択し、解凍した 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.zip が \python_demo-master\ に表示されると、パッケージの準備は完了です。

UDSFの登録

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

UDSFの使用方法

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

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

   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] > [Deployments] を選択します。対象のデプロイメントを見つけ、その [Actions] 列にある [Start] をクリックします。デプロイメントが開始されると、ASI_UDSF_Source の a フィールドのインデックス 2 から 4 の直前まで (インデックス 2 と 3) の文字が ASI_UDSF_Sink に書き込まれます。

Pythonインラインスカラー関数

インライン関数は、その実装を CREATE FUNCTION 文に直接埋め込むため、関数は同じ SQL 文内で定義および登録されます。次の例では、メールアドレスをマスクするインライン関数を定義します。$$ 区切り文字の間に完全なコードロジックを記述します。

CREATE TEMPORARY FUNCTION mask_email(email STRING)
RETURNS STRING AS $$
if email is None:
    return None

name, separator, domain = email.partition("@")
if not separator:
    return email

return name[:1] + "***@" + domain
$$ LANGUAGE PYTHON;

非同期ユーザー定義関数

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

import asyncio

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


# 方法1: Pythonの非同期関数を使用
@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()
)