Realtime Compute for Apache Flink:PyFlink ジョブの開発

Table API／SQL と DataStream API の選択

PyFlink では Table API／SQL および DataStream API の両方がサポートされています。Table API／SQL の使用を推奨します。その理由は以下のとおりです：

• 優れたパフォーマンス：Table API／SQL の実行計画は最適化後に JVM 内で完全に実行されます。一方、DataStream API では JVM と Python プロセス間でレコード単位のシリアル化およびデシリアル化が発生するため、大きなオーバーヘッドが発生します。

• より豊富な機能：Table API／SQL では、コネクタ、データ形式、ウィンドウ関数などに対する完全なサポートが提供されます。また、SQL ジョブと同一のコネクタエコシステムを共有します。

• コミュニティによる推奨：Apache Flink コミュニティでは、PyFlink 開発において Table API／SQL が優先されています。

SQL で複雑なカスタムロジックを表現できない場合にのみ、DataStream API を使用してください。

開発の参考情報

以下に示すリソースを活用して、Flink ビジネスロジックをローカルで開発してください。開発完了後、コードを Realtime Compute for Apache Flink コンソールにアップロードし、ジョブをデプロイします。

• Apache Flink 1.20 におけるビジネスロジックの開発方法については、「Flink Python API 開発ガイド」をご参照ください。

• Apache Flink のコーディング中に発生する一般的な問題とその解決方法については、「よくある質問」をご参照ください。

プロジェクト構造

推奨される Python ジョブプロジェクトの構造は以下のとおりです：

my-flink-python-project/
├── my_job.py                # メインジョブファイル
├── udfs.py                  # ユーザー定義関数（任意）
├── requirements.txt         # サードパーティ製 Python 依存関係（任意）
└── config.properties        # 構成ファイル（任意）

依存関係の管理

PyFlink ジョブでカスタム Python 仮想環境、サードパーティ製 Python パッケージ、JAR、およびデータファイルを使用する方法については、「Python 依存関係の使用」をご参照ください。

ユーザー定義関数（UDF）

以下の例では、文字列を匿名化する Python UDF（ユーザー定義関数）の開発方法を示します：

from pyflink.table import DataTypes
from pyflink.table.udf import udf

@udf(result_type=DataTypes.STRING())
def mask_phone(phone: str):
    """電話番号をマスク：先頭 3 桁と末尾 4 桁を保持し、中間の桁を **** で置き換えます。"""
    if phone is None or len(phone) != 11:
        return phone
    return phone[:3] + '****' + phone[7:]

この UDF を SQL ジョブで使用するには、以下のとおりです：

CREATE TEMPORARY FUNCTION mask_phone AS 'udfs.mask_phone' LANGUAGE PYTHON;
INSERT INTO sink_table
SELECT name, mask_phone(phone) AS masked_phone
FROM source_table;

UDF の登録、更新、削除方法については、「ユーザー定義関数（UDF）の管理」をご参照ください。

コネクタの使用

Flink がサポートするコネクタの一覧については、「サポートされているコネクタ」をご参照ください。コネクタを使用するには、以下の手順を実行します：

1. Realtime Compute for Apache Flink コンソール にログインします。

2. 対象のワークスペースの [操作] 列で、[コンソール] をクリックします。

3. 左側のナビゲーションウィンドウで、[ファイル管理] をクリックします。

4. [Upload Artifact] をクリックし、対象のコネクタの Python パッケージを選択します。

   Realtime Compute for Apache Flink から提供されるカスタムコネクタまたは組み込みコネクタをアップロードできます。Flink コネクタ向け公式 Python パッケージのダウンロードリンクについては、「コネクタ一覧」をご参照ください。

5. ［オペレーションセンター > ジョブ運用管理］ページで、［ジョブのデプロイ > Pythonジョブ］をクリックし、［追加の依存関係］フィールドで対象のコネクタ用のPythonパッケージを選択し、他のパラメーターを設定して、ジョブをデプロイします。

6. デプロイ済みのジョブの名前をクリックします。[デプロイメントの詳細] タブの [実行時パラメーター設定] セクションで、[編集] をクリックし、[その他の設定] に Python コネクタパッケージへのパスを追加します。

   ジョブが複数のコネクタパッケージ（例：connector-1.jar、connector-2.jar）に依存する場合、以下のように設定します：

   pipeline.classpaths: 'file:///flink/usrlib/connector-1.jar;file:///flink/usrlib/connector-2.jar'

7. 組み込みコネクタ、データ形式、またはカタログ（VVR 11.2 以降でのみ利用可能）を使用する場合は、**[パラメーター]** の下にある **[その他の構成]** フィールドに構成を追加します。例：

   ## 複数の組み込みコネクタを使用
   pipeline.used-builtin-connectors: kafka;sls
   ## 複数のデータ形式を使用
   pipeline.used-builtin-formats: avro;parquet
   ## 複数の既存カタログを使用
   pipeline.used-builtin-catalogs: catalogname1;catalogname2

コネクタの使用に関する詳細なサンプルコードについては、「完全なサンプルコード」をご参照ください。

ジョブのデバッグ

import logging

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

ローカルデバッグ

デフォルトでは、Realtime Compute for Apache Flink はインターネットにアクセスできません。そのため、テスト目的でオンラインデータソースに直接接続できない場合があります。ローカルデバッグには、以下のアプローチを推奨します：

• 単体テスト：UDF のロジックの正しさを検証するために、独立した単体テストを実行します。

• ローカル実行：ファイルやメモリ内データなどのローカルデータソースを用いて入力をシミュレートし、ジョブをローカルで実行して処理ロジックを検証します。例：

  from pyflink.datastream import StreamExecutionEnvironment
  
  env = StreamExecutionEnvironment.get_execution_environment()
  # ローカルデータソースを使用したテスト
  ds = env.from_collection([('Alice', 1), ('Bob', 2), ('Alice', 3)])
  ds.key_by(lambda x: x[0]).sum(1).print()
  env.execute("local_test")

• リモートデバッグ：オンラインデータソースを用いたデバッグについては、「コネクタを用いたジョブのローカル実行およびデバッグ」をご参照ください。

ジョブのデプロイメント

PyFlink ジョブの開発が完了したら、Realtime Compute for Apache Flink コンソールにアップロードしてデプロイします。以下の手順に従ってください：

1. Realtime Compute for Apache Flink コンソール にログインし、対象のワークスペースに移動します。

2. 左側のナビゲーションウィンドウで、ファイル管理 をクリックし、Python ジョブファイル（.py または .zip）をアップロードします。また、サードパーティ製の依存関係や構成ファイルも必要に応じてアップロードします。

3. オペレーションセンター > ジョブ O&M ページで、ジョブのデプロイ > Python ジョブ をクリックし、デプロイ情報に入力します。

   パラメーター	説明
   Python ファイルパス	アップロード済みの Python ジョブファイルを選択します。
   エントリーモジュール	.py ファイルの場合は空欄のままにしてください。.zip ファイルの場合は、エントリーモジュール名（例：my_job）を入力します。
   追加の依存関係	必要に応じて、コネクタ JAR や構成ファイルを選択します。
   Python ライブラリー	必要に応じて、サードパーティ製 Python パッケージ（.whl または .zip）を選択します。
   Python アーカイブ	必要に応じて、カスタム Python 仮想環境（.zip）を選択します。

4. デプロイ をクリックします。

デプロイパラメーターの詳細については、「ジョブのデプロイメント」をご参照ください。