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

Table API/SQL と DataStream API の選択

PyFlink では、Table API/SQL および DataStream API の 2 つの開発アプローチがサポートされています。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 ビジネスコードをローカルで開発する際は、以下のドキュメントを参照できます。開発が完了したら、コードを Flink 開発コンソールにアップロードしてジョブをデプロイします。

• Apache Flink V1.20 のビジネスコード開発については、「Flink Python API 開発ガイド」をご参照ください。

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

プロジェクト構造

Python ジョブの推奨プロジェクト構造は次のとおりです。

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

依存関係管理

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

ユーザー定義関数 (UDF)

以下は、機密文字列データをマスクする Python UDSF の開発例です。

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 コンソール にログインします。

2. 対象のワークスペースの Actions 列で Console をクリックします。

3. 左側のナビゲーションウィンドウで File Management をクリックします。

4. Upload Resource をクリックし、対象コネクタの Python パッケージを選択してアップロードします。

   独自開発のコネクタまたは Flink が提供するコネクタをアップロードできます。Flink が提供するコネクタの公式 Python パッケージをダウンロードするには、「コネクタ一覧」をご参照ください。

5. O&M Center > Job O&M ページで、Create Deployment > Python Deployment をクリックします。Additional Dependencies フィールドで対象コネクタの Python パッケージを選択し、他のパラメーターを設定してジョブをデプロイします。

6. デプロイされたジョブの名前をクリックします。Deployment Details タブの Running Parameters セクションで Edit をクリックします。Additional Configuration に Python コネクタパッケージの場所情報を追加します。

   ジョブが複数のコネクタ Python パッケージに依存する場合（たとえば、connector-1.jar および connector-2.jar という 2 つのパッケージ）、構成は次のようになります。

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

7. ビルトインコネクタ、データ形式、およびカタログ（VVR 11.2 以降のみ）を使用するには、ジョブの Running Parameters セクションの Additional Configuration に構成を追加します。例を以下に示します。

   ## 複数のコネクタを使用
   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")

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

ジョブデプロイメント

Python ジョブの開発が完了したら、ジョブを Realtime Compute コンソールにアップロードしてデプロイします。次の手順を実行してください。

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

2. 左側のナビゲーションウィンドウで File Management をクリックし、Python ジョブファイル（.py または .zip）をアップロードします。サードパーティの依存関係や設定ファイルがある場合は、それらもアップロードしてください。

3. O&M Center > Job O&M ページで、Create Deployment > Python Deployment をクリックし、デプロイメント情報を入力します。

   パラメーター	説明
   Python File Path	アップロード済みの Python ジョブファイルを選択します。
   Entry Module	ジョブファイルが .py ファイルの場合、このフィールドは不要です。ジョブファイルが .zip ファイルの場合、エントリモジュール名（例：my_job）を指定します。
   Additional Dependencies	該当する場合は、コネクタ JAR ファイルまたは設定ファイルを選択します。
   Python Libraries	該当する場合は、サードパーティ Python パッケージ（.whl または .zip）を選択します。
   Python Archives	該当する場合は、カスタム Python 仮想環境（.zip）を選択します。

4. Deploy をクリックします。

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