Pythonパイプラインの依存関係の管理

依存関係管理とは、パイプラインに必要な依存関係を指定し、本番環境で使用する依存関係を制御することです。

注: パイプラインの実行に使用されるリモートワーカーは、通常、Debianベースのコンテナイメージに標準のPythonディストリビューションがインストールされています。コードが標準のPythonパッケージのみに依存している場合は、このページで何もする必要はないでしょう。

PyPIの依存関係

パイプラインがPython Package Indexの公開パッケージを使用する場合は、これらのパッケージをワーカー上でリモートで利用できるようにする必要があります。

単一のPythonファイルまたはノートブックのみで構成されるパイプラインの場合、依存関係を提供する最も簡単な方法は、requirements.txtファイルを提供することです。より複雑なシナリオでは、パッケージでパイプラインを定義し、カスタムコンテナに依存関係をインストールすることを検討してください。

requirements.txtファイルを提供するには

1. マシンにどのパッケージがインストールされているかを確認します。次のコマンドを実行します

    pip freeze > requirements.txt
   

   このコマンドは、インストール元に関係なく、マシンにインストールされているすべてのパッケージをリストするrequirements.txtファイルを作成します。

2. requirements.txtファイルを編集し、コードに関係のないすべてのパッケージを削除します。

3. 次のコマンドラインオプションを指定してパイプラインを実行します

    --requirements_file requirements.txt
   

   ランナーはrequirements.txtファイルを使用して、追加の依存関係をリモートワーカーにインストールします。

注: pip freezeの代わりに、pip-toolsのようなライブラリを使用して、requirements.inファイルからパイプラインに必要なすべての依存関係をコンパイルします。requirements.inファイルには、最上位の依存関係のみが記載されています。

--requirements_fileパイプラインオプションを指定すると、パイプラインの送信中に、Beamは指定されたパッケージをローカルのrequirementsキャッシュディレクトリにダウンロードし、requirementsキャッシュディレクトリをランナーにステージングします。実行時に、利用可能な場合、Beamはrequirementsキャッシュからパッケージをインストールします。このメカニズムにより、送信時に依存関係パッケージをランナーにステージングすることが可能になります。実行時に、ランナーワーカーはPyPIへの接続を必要とせずに、キャッシュからパッケージをインストールできる場合があります。requirementsのステージングを無効にするには、--requirements_cache=skipパイプラインオプションを使用します。詳細については、これらのパイプラインオプションのヘルプ説明を参照してください。

カスタムコンテナ

パイプラインに必要なすべての依存関係を含むコンテナイメージを渡すことができます。カスタムコンテナイメージを使用してパイプラインを実行する方法を示す手順に従ってください。

1. カスタムコンテナイメージを使用している場合は、ビルド時に--requirements_fileから依存関係をイメージに直接インストールすることをお勧めします。この場合、実行時に--requirements_fileオプションを渡す必要がないため、パイプラインの起動時間が短縮されます。

   # Add these lines with the path to the requirements.txt to the Dockerfile
   COPY <path to requirements.txt> /tmp/requirements.txt
   RUN python -m pip install -r /tmp/requirements.txt
   

ローカルPythonパッケージまたは非公開Python依存関係

パイプラインが(GitHubリポジトリからダウンロードしたパッケージなど)公開されていないパッケージを使用する場合は、次の手順を実行して、これらのパッケージをリモートで利用できるようにします。

1. マシンにインストールされていて、公開されていないパッケージを特定します。次のコマンドを実行します

   pip freeze

   このコマンドは、インストール元に関係なく、マシンにインストールされているすべてのパッケージをリストします。

   1. 次のコマンドラインオプションを指定してパイプラインを実行します

       --extra_package /path/to/package/package-name
      

      ここで、package-nameはパッケージのtarボールです。buildというコマンドラインツールを使用して、パッケージのtarボールを作成できます。

        # Install build using pip
        pip install --upgrade build
        python -m build --sdist
      

      このコマンドの詳細については、buildドキュメントを参照してください。

複数ファイルの依存関係

多くの場合、パイプラインコードは複数のファイルにまたがります。プロジェクトをリモートで実行するには、これらのファイルをPythonパッケージとしてグループ化し、パイプラインの実行時にパッケージを指定する必要があります。リモートワーカーが起動すると、パッケージがインストールされます。ファイルをPythonパッケージとしてグループ化し、リモートで利用できるようにするには、次の手順を実行します

1. プロジェクトのsetup.pyファイルを作成します。以下は非常に基本的なsetup.pyファイルです。

    import setuptools
   
    setuptools.setup(
       name='PACKAGE-NAME',
       version='PACKAGE-VERSION',
       install_requires=[
         # List Python packages your pipeline depends on.
       ],
       packages=setuptools.find_packages(),
    )
   

2. ルートディレクトリにsetup.pyファイル、メインワークフローファイル、および残りのファイルを含むディレクトリが含まれるようにプロジェクトを構成します。たとえば、

    root_dir/
      setup.py
      main.py
      my_package/
        my_pipeline_launcher.py
        my_custom_dofns_and_transforms.py
        other_utils_and_helpers.py
   

   このプロジェクト構造に従った例については、Juliasetを参照してください。

3. たとえば、次のコマンドを使用して、送信環境にパッケージをインストールします

    pip install -e .
   

4. 次のコマンドラインオプションを指定してパイプラインを実行します

    --setup_file /path/to/setup.py
   

注: パッケージの依存関係がsetup.pyファイルのinstall_requiresフィールドで定義されている場合(手順1を参照)、--requirements_fileオプションを指定する必要はありません。ただし、--requirements_fileオプションとは異なり、--setup_fileオプションを使用すると、Beamは依存パッケージをランナーにステージングしません。パイプラインパッケージのみがステージングされます。それらがランタイム環境でまだ提供されていない場合は、パッケージの依存関係が実行時にPyPIからインストールされます。

Python以外の依存関係またはPython以外の依存関係を持つPyPIの依存関係

apt installコマンドを使用してインストールする必要があるパッケージなど、Python以外のパッケージを使用する場合、またはパッケージのインストール中にPython以外の依存関係に依存するPyPIパッケージを使用する場合は、カスタムコンテナを使用してインストールすることをお勧めします。それ以外の場合は、次の手順を実行する必要があります。

1. パイプラインをパッケージとして構成します.

2. apt installコマンドなどのPython以外の依存関係に必要なインストールコマンドを、setup.pyファイルのCUSTOM_COMMANDSリストに追加します。例については、Juliaset setup.pyファイルを参照してください。

   注: これらのコマンドがリモートワーカーで実行されることを確認する必要があります。たとえば、aptを使用する場合は、リモートワーカーにaptサポートが必要です。

3. 次のコマンドラインオプションを指定してパイプラインを実行します

    --setup_file /path/to/setup.py
   

注: カスタムコマンドは、ワークフローの依存関係が(pipによって)インストールされた後に実行されるため、パイプラインのrequirements.txtファイルと、setup.pyファイルのsetuptools.setup()呼び出しのinstall_requiresパラメーターからPyPIパッケージの依存関係を省略する必要があります。

SDKコンテナイメージの事前構築

BeamランナーがDockerコンテナ内でSDKワーカーを起動するパイプライン実行モードでは、追加のパイプライン依存関係（--requirements_fileやその他のランタイムオプションで指定）は、実行時にコンテナにインストールされます。これにより、ワーカーの起動時間が長くなる可能性があります。ただし、--prebuild_sdk_container_engineを使用すると、SDKコンテナを事前にビルドし、ワーカーが起動する前に一度だけ依存関係のインストールを実行できる場合があります。Google Cloud Dataflowでの事前ビルドの使用方法については、追加の依存関係を含むPython SDKカスタムコンテナイメージの事前ビルドを参照してください。

注記: この機能はDataflow Runner v2でのみ利用可能です。

メインセッションのピクル化と管理

Python SDKがリモートランナーに実行のためにパイプラインを送信する際、変換ユーザーコードなどのパイプラインの内容は、シリアライズ（またはピクル化）を実行するライブラリ（ピクラーとも呼ばれる）を使用して、バイトコードにシリアライズされます。Beamで使用されるデフォルトのピクラーライブラリはdillです。cloudpickleピクラーを使用するには、--pickle_library=cloudpickleパイプラインオプションを指定します。cloudpickleのサポートは現在実験的です。

デフォルトでは、メインパイプラインモジュールで定義されたグローバルなインポート、関数、および変数は、Beamジョブのシリアライズ中に保存されません。そのため、リモートランナーでDoFnを実行する際に、予期しないNameErrorが発生する可能性があります。これを解決するには、--save_main_sessionパイプラインオプションを設定して、メインセッションの内容をパイプラインに供給します。これにより、グローバル名前空間のピクル化された状態がDataflowワーカー（DataflowRunnerを使用している場合）にロードされます。たとえば、DataflowRunnerでメインセッションを設定するには、NameErrorの処理を参照してください。

Python SDKでメインセッションを管理する必要があるのは、リモートランナーでdillピクラーを使用する場合のみです。したがって、この問題はDirectRunnerでは発生しません。

パイプラインのシリアライズはジョブの送信時に行われ、デシリアライズは実行時に行われるため、ジョブの送信時と実行時に同じバージョンのピクル化ライブラリを使用することが不可欠です。これを確実にするために、Beamは通常、ピクル化ライブラリに対して非常に狭いサポート対象バージョン範囲を設定しています。何らかの理由で、ユーザーがBeamで要求されるdillまたはcloudpickleのバージョンを使用できず、カスタムバージョンをインストールすることを選択した場合、実行時（例えばカスタムコンテナ内やパイプライン依存関係要件の指定）でも同じカスタムバージョンを使用する必要があります。

パイプラインが使用する依存関係の制御

パイプライン環境

リモートランナーでPythonパイプラインを実行するために、Apache Beamはパイプラインをランナーに依存しない表現に変換し、実行のために送信します。変換は起動環境で行われます。Beam SDKがインストールされたPython仮想環境、またはDataflow Flex Templates、ノートブック環境、Apache Airflowなどのツールを使用して、パイプラインを起動できます。

ランタイム環境は、パイプラインの実行中にランナーが使用するPython環境です。この環境は、データ処理を実行する際にパイプラインコードが実行される場所です。ランタイム環境には、Apache Beamとパイプラインのランタイム依存関係が含まれます。

再現可能な環境を作成する

再現可能なPython環境を構築するために、いくつかのツールを使用できます。

• 要件ファイルを使用します。依存関係をインストールした後、pip freeze > requirements.txtを使用して要件ファイルを生成します。環境を再作成するには、pip install -r requirements.txtを使用してrequirements.txtファイルから依存関係をインストールします。

• 制約ファイルを使用します。制約リストを使用してパッケージのインストールを制限し、指定されたバージョンのみを許可できます。

• ロックファイルを使用します。PipEnv、Poetry、pip-toolsなどの依存関係管理ツールを使用して、トップレベルの依存関係を指定し、ピン留めされたバージョンを持つすべての推移的な依存関係のロックファイルを生成し、これらのロックファイルから仮想環境を作成します。

• Dockerコンテナイメージを使用します。Dockerコンテナイメージ内に起動環境とランタイム環境をパッケージ化できます。イメージにすべての必要な依存関係が含まれている場合、環境はコンテナイメージが再構築された場合にのみ変更されます。

環境を定義する構成ファイルにはバージョン管理を使用します。

パイプラインのランタイム環境を再現可能にする

パイプラインがリモートランナーで再現可能なランタイム環境を使用する場合、ランナーのワーカーはパイプラインが実行されるたびに同じ依存関係を使用します。再現可能な環境は、パイプラインの直接または推移的な依存関係のリリースによって引き起こされる副作用の影響を受けません。実行時に依存関係の解決は必要ありません。

次の方法で再現可能なランタイム環境を作成できます。

• パイプラインのすべての依存関係を持つカスタムコンテナイメージでパイプラインを実行します。--sdk_container_imageパイプラインオプションを使用します。

• --requirements_fileパイプラインオプションで、パイプラインの依存関係の完全なリストを提供します。パイプライン実行前にランタイム環境の初期化シーケンスを実行するには、--prebuild_sdk_container_engineオプションを使用します。依存関係が変更されない場合は、--sdk_container_imageオプションを使用して、事前ビルドされたイメージを再利用します。

自己完結型のランタイム環境は通常、再現可能です。ランタイム環境が自己完結型かどうかを確認するには、パイプラインランタイムでPyPIへのインターネットアクセスを制限します。Dataflowランナーを使用する場合は、--no_use_public_ipsパイプラインオプションのドキュメントを参照してください。

ランタイム環境を再作成またはアップグレードする必要がある場合は、変更された依存関係を可視化しながら制御された方法で実行します。

• パイプラインが実行中に、コンテナイメージを変更しないでください。

• カスタムイメージで:latestタグを使用しないようにしてください。日付または一意の識別子でビルドにタグを付けます。何かがうまくいかない場合、このタイプのタグを使用すると、パイプラインの実行を以前に既知の動作する構成に戻して、変更を検査することが可能になる場合があります。

• pip freezeの出力またはrequirements.txtの内容をバージョン管理システムに保存することを検討してください。

パイプラインの起動環境を再現可能にする

起動環境はパイプラインの本番バージョンを実行します。パイプラインをローカルで開発している間は、JupyterやPylintなどの開発用依存関係を含む開発環境を使用する場合があります。本番パイプラインの起動環境には、これらの追加の依存関係は必要ない場合があります。開発環境とは別に構築および維持できます。

パイプラインの送信に対する副作用を減らすには、再現可能な方法で起動環境を再作成できるようにするのが最善です。

Dataflow Flex Templatesは、コンテナ化された再現可能な起動環境の例を提供します。

クリーンな仮想環境へのBeamの再現可能なインストールを作成するには、Beamのデフォルトコンテナイメージの制約ファイルに含まれるすべてのPython依存関係をリストする要件ファイルを使用します。

BEAM_VERSION=2.48.0
PYTHON_VERSION=`python -c "import sys; print(f'{sys.version_info.major}{sys.version_info.minor}')"`
pip install apache-beam==$BEAM_VERSION --constraint https://raw.githubusercontent.com/apache/beam/release-${BEAM_VERSION}/sdks/python/container/py${PY_VERSION}/base_image_requirements.txt

制約ファイルを使用して、起動環境のBeamの依存関係がデフォルトのBeamコンテナのバージョンと一致することを確認します。制約ファイルは、インストール時の依存関係解決の必要性をなくすこともあります。

起動環境とランタイム環境の互換性を確保する

起動環境は、パイプライングラフをランナーに依存しない表現に変換します。このプロセスには、変換のコードのシリアライズ（またはピクル化）が含まれます。シリアライズされた内容はワーカーでデシリアライズされます。ランタイムワーカー環境が起動環境と大幅に異なる場合、次の理由でランタイムエラーが発生する可能性があります。

• Apache Beamのバージョンは、送信環境とランタイム環境で一致する必要があります。Pythonのメジャーバージョンとマイナーバージョンも一致する必要があります。そうしないと、パイプラインがパイプライン構築環境とパイプラインランタイム環境は互換性がありませんのようなエラーで失敗する可能性があります。古いSDKバージョンでは、エラーはSystemError: unknown opcodeとして報告される場合があります。

• 送信環境とランタイム環境のprotobufのバージョンは、一致するか互換性がある必要があります。

• パイプラインコードで使用されるライブラリは一致する必要がある場合があります。シリアライズされたパイプラインコードに、ワーカーで使用できない関数またはモジュールへの参照がある場合、パイプラインはリモートランナーでModuleNotFoundまたはAttributeError例外で失敗する可能性があります。このようなエラーが発生した場合は、影響を受けるライブラリがリモートワーカーで利用可能であることを確認し、メインセッションを保存する必要があるかどうかを確認してください。

• 送信時に使用されるピクル化ライブラリのバージョンは、実行時にインストールされたバージョンと一致する必要があります。これを強制するために、Beamはシリアライザーライブラリ（dillとcloudpickle）のバージョンに厳しい制約を設定しています。次の条件下では、Beamで要求されるものとは異なるバージョンのdillまたはcloudpickleを強制的にインストールできます。

  • 送信環境とランタイム環境に同じバージョンをインストールします。
  • 選択したバージョンがパイプラインで機能します。

ランタイム環境が起動環境と一致するかどうかを確認するには、両方の環境のpip freeze出力の違いを調べます。環境互換性チェックが新しいSDKバージョンに含まれているため、Beamの最新バージョンに更新します。

最後に、ランタイムで使用するコンテナ化された環境からパイプラインを起動することで、同じ環境を使用できます。カスタムコンテナイメージから構築されたDataflow Flexテンプレートは、この設定を提供します。このシナリオでは、起動環境とランタイム環境の両方を再現可能な方法で再作成できます。両方のコンテナは同じイメージから作成されるため、起動環境とランタイム環境はデフォルトで互換性があります。