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

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 に集中させています。

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

開発リファレンス

Flink のビジネスコードをローカルで開発するには、以下のドキュメントを使用してください。開発後、コードを 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        # 設定ファイル (オプション)

依存関係の管理

カスタム Python 仮想環境、サードパーティの Python パッケージ、JAR、およびデータファイルを Python ジョブで使用する方法については、「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) の管理」をご参照ください。

コネクタの使用

サポートされているコネクタのリストについては、「サポートされているコネクタ」をご参照ください。コネクタを使用するには、次の手順に従ってください：

1. リアルタイムコンピューティングコンソールにログインします。

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

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

4. [アーティファクトのアップロード] をクリックし、ターゲットコネクタの Python パッケージを選択します。

   カスタムコネクタまたは組み込みの Flink コネクタをアップロードできます。公式 Flink コネクタの Python パッケージのダウンロードリンクについては、「コネクタリスト」をご参照ください。

5. [O&M] > [ジョブ O&M] ページで、[ジョブのデプロイ] > [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

ジョブのデバッグ

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 ジョブを開発した後、リアルタイムコンピューティングコンソールにアップロードしてデプロイします。次の手順に従ってください：

1. リアルタイムコンピューティングコンソールにログインし、対象のワークスペースに移動します。

2. 左側のナビゲーションウィンドウで、[ファイル管理] をクリックします。Python ジョブファイル (.py または .zip) を、サードパーティの依存関係や設定ファイルとともにアップロードします。

3. [O&M] > [ジョブ O&M] ページで、[ジョブのデプロイ] > [Python ジョブ] をクリックし、デプロイ情報を入力します。

   パラメーター	説明
   Python ファイルパス	アップロードした Python ジョブファイルを選択します。
   エントリモジュール	ジョブファイルが .py ファイルの場合は空白のままにします。.zip ファイルの場合は、my_job のようにエントリモジュール名を入力します。
   追加の依存関係	コネクタの JAR や設定ファイルをここで選択します。
   Python ライブラリ	サードパーティの Python パッケージ (.whl または .zip) をここで選択します。
   Python アーカイブ	カスタム Python 仮想環境 (.zip) をここで選択します。

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

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