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

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

Flink 完全管理環境には、Python 環境がプリインストールされています。 Python のバージョンは次のとおりです。

• Ververica Runtime (VVR) 8.0.10 以前： Python 3.7

• VVR 8.0.11 以降： Python 3.9

一部のサードパーティ 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 バージョンに変更します。 詳細については、「ストレージ管理」をご参照ください。

   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] タブで、管理するワークスペースを見つけ、[Actions] 列の [Console] をクリックします。

   3. 左側メニューで、[アーティファクト] をクリックします。 [アーティファクト] ページで、[Upload Artifact] をクリックします。 表示されるダイアログボックスで、venv.zip パッケージを選択します。

   4. 運用保守 > デプロイメント ページで、目的のジョブの名前をクリックします。

   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. [Fully Managed Flink] タブで、管理するワークスペースを見つけ、[Actions] 列の [Console] をクリックします。

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

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

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

   たとえば、ドラフトが jar1.jar と jar2.jar という名前の2つの JAR パッケージに依存している場合は、次の設定を追加します。

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

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

組み込みコネクタ、データフォーマット、カタログの使用

Python プログラムで組み込みのコネクタ、データフォーマット、カタログを使用するには、次の手順を実行してください。

1. Python デプロイメントの詳細ページで、[パラメーター] セクションの [その他の設定] フィールドに設定を追加します。

   組み込みコネクタを使用するためのパラメーターを追加します。 次の設定では、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. [Save] をクリックします。

データファイルの使用

関連ドキュメント

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

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

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