Realtime Compute for Apache Flink:Python UDTF

UDTF の開発

注記

Flink は、UDX の開発を支援するために、Python ユーザー定義拡張 (UDX) のサンプルプロジェクトを提供しています。サンプルには、Python ユーザー定義スカラー関数、ユーザー定義集計関数 (UDAF)、および UDTF の実装が含まれています。この例では、Windows オペレーティングシステムを使用して UDTF を開発する方法について説明します。

重要

• 登録する UDF 名には、小文字、数字、ハイフン (-) のみを使用できます。アンダースコア (_) やその他の特殊文字は使用できません。

• Realtime Compute for Apache Flink のマネージド環境には、split のような組み込み関数がいくつかプリインストールされています。カスタム関数が組み込み関数と同じ名前の場合、登録に失敗するか、SQL 検証でエラーが発生します。関数名には、my-split のようなカスタムプレフィックスを使用してください。

1. python_demo-master サンプルパッケージをダウンロードし、コンピューターに解凍します。

2. PyCharm で、[file] > [open] を選択して、解凍した python_demo-master フォルダーを開きます。

3. \python_demo-master\udx\udtfs.py をダブルクリックし、ビジネス要件に基づいて udtfs.py ファイルの内容を変更します。

この例では、my_split は各行の文字列を縦棒 (|) で複数の列に分割します。

    from pyflink.table import DataTypes
    from pyflink.table.udf import udtf
    
    @udtf(result_types=[DataTypes.STRING(), DataTypes.STRING()])
    def my_split(s: str):
        splits = s.split("|")
        yield splits[0], splits[1]
        

1. ダウンロードしたパッケージ (\python_demo) 内の udx フォルダーが含まれているディレクトリで、以下のコマンドを実行してファイルをパッケージ化します。

    zip -r python_demo.zip udx
        

\python_demo\ ディレクトリに python_demo.zip パッケージが生成されると、UDTF の開発は完了です。

UDTF の登録

UDTF の登録方法の詳細については、「UDF の管理」をご参照ください。

UDTF の使用

UDTF が登録されると、使用できるようになります。以下の手順を実行してください。

1. Flink SQL ジョブを開発します。詳細については、「ジョブ開発の概要」をご参照ください。

ASI_UDTF_Source テーブルの各行の message フィールドの値は、文字列 aa と縦棒 (|) で連結された後、さらに縦棒 (|) で複数の列に分割されます。

    CREATE TEMPORARY TABLE ASI_UDTF_Source (
      `message`  VARCHAR
    ) WITH (
      'connector'='datagen'
    );
    
    CREATE TEMPORARY TABLE ASI_UDTF_Sink (
      name  VARCHAR,
      place  VARCHAR
    ) WITH (
      'connector' = 'blackhole'
    );
    
    INSERT INTO ASI_UDTF_Sink
    SELECT name,place
    FROM ASI_UDTF_Source,lateral table(my_split(concat_ws('|', `message`, 'aa'))) as T(name,place);
        

1. [O&M] > [Deployments] ページで、対象ジョブの [Actions] 列にある [Start] をクリックします。

ジョブが開始されると、ASI_UDTF_Sink テーブルには 2 つの列が含まれます。このデータは、ASI_UDTF_Source テーブルの各行の message フィールドと aa フィールドを縦棒 (|) で連結し、その結果を縦棒 (|) で分割して生成されます。