Realtime Compute for Apache Flink:Python 依存関係の使用

プリインストールされた Python 環境

フルマネージド Flink 環境には、Python 環境がプリインストールされています。Python のバージョンは以下のとおりです。

• VVR 8.0.10 以前：Python 3.7

• VVR 8.0.11 以降、VVR 11.7 より前：Python 3.9

• VVR 11.7 以降：Python 3.9、Python 3.10、および Python 3.11

一部のサードパーティの Python パッケージには glibc バージョンの要件があります。フルマネージド Flink 環境にプリインストールされている glibc のバージョンは以下のとおりです。

カスタム Python 仮想環境の使用

プリインストールされた Python 環境が要件を満たさない場合は、Python 仮想環境を使用してカスタム Python バージョンを使用できます。各 Python 仮想環境は、完全な Python ランタイム環境を提供します。仮想環境には、一連の Python 依存関係パッケージをインストールできます。以下のセクションでは、Python 仮想環境を準備する方法について説明します。

1. Python 仮想環境を準備します。

   1. ローカルデバイスに setup-pyflink-virtual-env.sh スクリプトを用意します。 スクリプトの内容は以下のとおりです。

      X86

      set -e
      # miniforge.sh スクリプトをダウンロードします。
      wget "https://github.com/conda-forge/miniforge/releases/download/25.11.0-1/Miniforge3-25.11.0-1-Linux-x86_64.sh" -O "miniforge.sh"
      
      # miniforge.sh スクリプトに実行権限を追加します。
      chmod +x miniforge.sh
      
      # miniforge をインストールします。
      ./miniforge.sh -b
      source /root/miniforge3/bin/activate
      
      # Python 仮想環境を作成します。
      mamba create -n venv python=3.10 -y
      eval "$(mamba shell hook --shell bash)"
      
      # Python 仮想環境をアクティブ化します。
      mamba activate venv
      
      # PyFlink 依存関係をインストールします。
      pip install "ververica-flink==11.7.0"
      # VVR 11.5 以前の場合、専用の PyPI パッケージは利用できません。代わりにオープンソースの PyFlink をインストールしてください：
      # pip install "apache-flink==1.20.3" "setuptools<81"
      
      # 不要な JAR ファイルを削除してパッケージサイズを削減します。
      find /root/miniforge3/envs/venv/lib/python3.10/site-packages/pyflink/ -name *.jar | xargs rm
      
      # Conda Python 仮想環境を非アクティブ化します。
      mamba deactivate
      
      # 準備した Conda Python 仮想環境をパッケージ化します。
      cd /root/miniforge3/envs/ && zip -r /root/venv.zip venv 

      ARM

      set -e
      # miniforge.sh スクリプトをダウンロードします。
      wget "https://github.com/conda-forge/miniforge/releases/download/25.11.0-1/Miniforge3-25.11.0-1-Linux-aarch64.sh" -O "miniforge.sh"
      
      # miniforge.sh スクリプトに実行権限を追加します。
      chmod +x miniforge.sh
      
      # miniforge をインストールします。
      ./miniforge.sh -b
      source /root/miniforge3/bin/activate
      
      # Python 仮想環境を作成します。
      mamba create -n venv python=3.10 -y
      eval "$(mamba shell hook --shell bash)"
      
      # Python 仮想環境をアクティブ化します。
      mamba activate venv
      
      # PyFlink 依存関係をインストールします。
      yum install -y java-11-openjdk-devel
      export JAVA_HOME=/usr/lib/jvm/java-11
      wget "https://raw.githubusercontent.com/apache/flink/release-1.20/flink-python/dev/dev-requirements.txt" -O dev-requirements.txt
      pip install -r dev-requirements.txt
      pip install "ververica-flink==11.7.0"
      # VVR 11.5 以前の場合、専用の PyPI パッケージは利用できません。代わりにオープンソースの PyFlink をインストールしてください：
      # pip install "apache-flink==1.20.3" "setuptools<81"
      
      # 不要な JAR ファイルを削除してパッケージサイズを削減します。
      find /root/miniforge3/envs/venv/lib/python3.10/site-packages/pyflink/ -name *.jar | xargs rm
      
      # Conda Python 仮想環境を非アクティブ化します。
      mamba deactivate
      
      # 準備した Conda Python 仮想環境をパッケージ化します。
      cd /root/miniforge3/envs && zip -r /root/venv.zip venv

      説明

      このトピックでは、デプロイメントで VVR 11.7 を使用し、Python 3.10 の仮想環境で実行します。別の VVR バージョンを使用する場合、または別の Python バージョンの仮想環境をインストールする場合は、以下のパラメータを変更する必要があります。

      • mamba create：希望する Python バージョンに変更します。

      • pip install：

        • VVR 11.6 以降：ververica-flink をインストールし、デプロイメントの VVR バージョンに合わせてバージョンを変更します。

        • VVR 11.5 以前：apache-flink をインストールし、デプロイメントの VVR バージョンに対応する Flink バージョンに変更します。Flink バージョンの確認方法の詳細については、「ストレージ管理」をご参照ください。

   2. ローカルデバイスで build.sh スクリプトを準備します。以下のコードはスクリプトの内容です。

      #!/bin/bash
      set -e -x
      
      yum install -y zip wget
      
      cd /root/
      bash /build/setup-pyflink-virtual-env.sh
      mv venv.zip /build/

   3. CLI で以下のコマンドを実行して、Python 仮想環境をインストールします。

      X86

      docker run -it --rm -v $PWD:/build -w /build quay.io/pypa/manylinux_2_28_x86_64 bash ./build.sh

      ARM

      docker run -it --rm -v $PWD:/build -w /build quay.io/pypa/manylinux_2_28_aarch64 bash ./build.sh

      コマンドを実行すると、venv.zip ファイルが生成されます。 この例では、Python 3.10 の仮想環境が使用されます。

      上記のスクリプトを変更して、仮想環境に必要なサードパーティの Python パッケージをインストールすることもできます。

2. Python デプロイメントで Python 仮想環境を使用します。

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

   2. Fully Managed Flink タブで、管理するワークスペースを見つけ、[アクション] 列の [コンソール] をクリックします。

   3. 左側のナビゲーションペインで、[アーティファクト] をクリックします。[アーティファクト] ページで、[アーティファクトのアップロード] をクリックします。表示されたダイアログボックスで、venv.zip パッケージを選択します。

   4. [O&M] > [デプロイメント] ページで、目的のジョブ名をクリックします。

   5. [設定] タブで、[基本] セクションの右上隅にある [編集] をクリックし、[Python アーカイブ] ドロップダウンリストから venv.zip パッケージを選択します。

      デプロイが Python ユーザー定義関数 (UDF) を使用する必要がある SQL デプロイの場合、[パラメーター] セクションの右上隅にある「編集」をクリックして、[その他の設定] フィールドに次の設定を追加します：

      python.archives: oss://.../venv.zip

   6. [パラメーター] セクションの [その他の設定] フィールドに、デプロイの VVR バージョンに応じた、指定の Python 仮想環境のインストールパスに関する設定情報を追加します。

      • VVR 6.X 以降

        python.executable: venv.zip/venv/bin/python
        python.client.executable: venv.zip/venv/bin/python

      • VVR 6.X より前のエンジンバージョン

        python.executable: venv.zip/venv/bin/python

サードパーティの Python パッケージの使用

JAR パッケージの使用

Python デプロイメントでコネクタや Java UDF などの Java クラスを使用する場合は、以下の手順で、コネクタまたは Java UDF の JAR パッケージを指定できます。

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

2. フルマネージド Flink タブで、管理するワークスペースを見つけ、[アクション] 列にある [コンソール] をクリックします。

3. 左側のナビゲーションペインで、[アーティファクト] をクリックします。[アーティファクト] ページで、[アーティファクトのアップロード] をクリックします。ダイアログボックスで、使用する JAR パッケージを選択します。

4. [デプロイ] ページで、目的のデプロイの名前をクリックします。 [設定] タブで、[基本] セクションの右上隅にある [編集] をクリックし、[追加の依存関係] ドロップダウンリストから必要な JAR パッケージを選択します。

5. [設定] タブで、[パラメーター] セクションの右上隅にある [編集] をクリックし、[その他の設定] フィールドに次の設定を追加します。

   たとえば、ドラフトが jar1.jar と jar2.jar の 2 つの JAR パッケージに依存する場合、以下の設定情報を追加します。

   pipeline.classpaths: 'file:///flink/usrlib/jar1.jar;file:///flink/usrlib/jar2.jar'

6. [保存] をクリックします。

組み込みコネクタ、データ形式、およびカタログの使用

Flink Python ジョブで組み込みコネクタ、データ形式、およびカタログを使用するには、以下の方法で指定します。

1. [パラメーター] セクションで、[その他の設定] フィールドに設定を追加します。

   組み込みコネクタの kafka および sls に依存するには、次の設定を追加します。 組み込みコネクタの名前については、サポートされているコネクタにある各コネクタのドキュメントをご参照ください。

   pipeline.used-builtin-connectors: kafka;sls

   avro と parquet の複数の組み込みデータ形式に依存するには、以下の設定を追加します。組み込みデータ形式の名前については、サポートされている データ形式 のドキュメントをご参照ください。

   pipeline.used-builtin-formats: avro;parquet

   hive-2.3.6 と paimon という名前の複数の組み込みカタログに依存するには、次の設定を追加します。 組み込みカタログの名前については、カタログにある各カタログのドキュメントをご参照ください。

   pipeline.used-builtin-catalogs: hive-2.3.6;paimon

2. [保存] をクリックします。

データファイルの使用

関連ドキュメント

• Python API ドラフトの開発方法の詳細については、「PyFlink ジョブの開発」をご参照ください。

• Realtime Compute for Apache Flink の Python デプロイメントの開発方法の詳細については、「PyFlink ジョブ」をご参照ください。

• フルマネージド Flink は SQL ドラフトと DataStream ドラフトをサポートします。SQL ドラフトと DataStream ドラフトの開発方法の詳細については、「ジョブ開発の概要」および「JAR ジョブの開発」をご参照ください。