Dataflow でパイプラインの依存関係を管理する

Apache Beam パイプラインの多くは、デフォルトの Dataflow ランタイム環境を使用して実行できます。ただし、一部のデータ処理のユースケースでは、追加のライブラリまたはクラスを使用することでメリットが得られます。このような場合には、パイプラインの依存関係を管理することが必要になる可能性があります。

パイプラインの依存関係の管理が必要になる理由には次のものがあります。

デフォルトのランタイム環境で提供される依存関係では、ユースケースに不十分である。
デフォルトの依存関係に、バージョンの競合が見られる、またはパイプラインコードと互換性のないクラスやライブラリが含まれている。
パイプラインで特定のライブラリバージョンに固定する必要がある。
一貫性のある一連の依存関係で実行する必要がある Python パイプラインが存在する。

依存関係の管理方法は、パイプラインで Java、Python、Go のいずれを使用するかによって異なります。

Java

クラスとライブラリに互換性がない場合、Java の依存関係に関する問題が発生する可能性があります。パイプラインにユーザー固有のコードと設定が含まれている場合、そのコードにライブラリの混合バージョンを含めることはできません。

Java の依存関係に関する問題

パイプラインで Java の依存関係に関する問題がある場合は、次のいずれかのエラーが発生する可能性があります。

NoClassDefFoundError: このエラーは、実行時にクラス全体を使用できない場合に発生します。
NoSuchMethodError: このエラーは、クラスパス内のクラスが正しいメソッドを含まないバージョンを使用している場合や、メソッドの署名が変更された場合に発生します。
NoSuchFieldError: このエラーは、クラスパスのクラスが、実行時に必要なフィールドがないバージョンを使用している場合に発生します。
FATAL ERROR: このエラーは、組み込み依存関係を正しく読み込めない場合に発生します。uber JAR ファイル（シェーディング済み）を使用する場合は、署名を使用するライブラリ（Conscrypt など）を同じ JAR ファイルに含めないでください。

依存関係の管理

Java パイプラインの依存関係管理を簡素化するために、Apache Beam は部品構成表（BOM）アーティファクトを使用します。BOM により、依存関係管理ツールで互換性のある依存関係の組み合わせを選択できます。詳細については、Apache Beam ドキュメントの Apache Beam SDK for Java の依存関係をご覧ください。

パイプラインで BOM を使用し、依存関係リストに他の依存関係を明示的に追加するには、SDK アーティファクトの pom.xml ファイルに次の情報を追加します。正しいライブラリ BOM をインポートするには、beam-sdks-java-io-google-cloud-platform-bom を使用します。

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-google-cloud-platform-bom</artifactId>
      <version>LATEST</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

<dependencies>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-core</artifactId>
  </dependency>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
  </dependency>
  <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
  </dependency>
</dependencies>

beam-sdks-java-core アーティファクトにはコア SDK だけが含まれています。I/O やランナーなどの他の依存関係は、依存関係リストに明示的に追加する必要があります。

Python

Apache Beam Python SDK を使用して Dataflow ジョブを実行する場合、依存関係の管理は次のシナリオで有効です。

パイプラインが Python Package Index（PiPy）の公開パッケージを使用しており、これらのパッケージをリモートで使用可能にする必要がある。
再現可能な環境を作成する必要がある。
起動時間を短縮するために、実行時にワーカーに依存関係がインストールされないようにする必要がある。

Python パイプラインの依存関係を定義する

1 つの Python スクリプトまたはノートブックを使用して Apache Beam パイプラインを作成できますが、Python エコシステムでは多くの場合、ソフトウェアはパッケージとして配布されます。パイプラインコードが複数のファイルに存在する場合は、パイプラインファイルを Python パッケージとしてグループ化すると、パイプラインの保守が容易になります。

パッケージの setup.py ファイルでパイプラインの依存関係を定義します。
--setup_file パイプラインオプションを使用して、パッケージをワーカーにステージングします。

リモートワーカーが起動するとパッケージがインストールされます。例については、Apache Beam GitHub の juliaset をご覧ください。

パイプラインを Python パッケージとして構造化するには、次の操作を行います。

プロジェクトの setup.py ファイルを作成します。setup.py ファイルに install_requires 引数を追加して、パイプラインの依存関係の最小セットを指定します。次の例は、基本的な setup.py ファイルを示しています。
```
import setuptools

setuptools.setup(
  name='PACKAGE_NAME',
  version='PACKAGE_VERSION',
  install_requires=[],
  packages=setuptools.find_packages(),
)
```
setup.py ファイル、メインワークフローファイル、残りのファイルを含むディレクトリをプロジェクトのルートディレクトリに追加します。このファイルグループは、パイプラインの Python パッケージです。ファイル構造は次の例のようになります。
```
root_dir/
  package_name/
    my_pipeline_launcher.py
    my_custom_transforms.py
    ...other files...
  setup.py
  main.py
```
パイプラインを実行するには、送信環境にパッケージをインストールします。パッケージをワーカーにステージングするには、--setup_file パイプラインオプションを使用します。次に例を示します。
```
python -m pip install -e .
python main.py --runner DataflowRunner --setup_file ./setup.py  <...other options...>
```

これらの手順により、特にコードのサイズと複雑さが増大した場合に、パイプラインコードのメンテナンスが簡素化されます。依存関係を指定するその他の方法については、Apache Beam ドキュメントの Python パイプラインの依存関係の管理をご覧ください。

カスタムコンテナを使用してランタイム環境を制御する

Apache Beam Python SDK を使用してパイプラインを実行するには、Dataflow ワーカーに、インタープリタ、Apache Beam SDK、パイプラインの依存関係を含む Python 環境が必要です。Docker コンテナイメージは、パイプラインコードを実行するための適切な環境を備えています。

ストックコンテナイメージは、Apache Beam SDK の各バージョンでリリースされます。これらのイメージには、Apache Beam SDK の依存関係が含まれています。詳細については、Apache Beam ドキュメントの Apache Beam SDK for Python の依存関係をご覧ください。

デフォルトのコンテナイメージに含まれていない依存関係がパイプラインで必要な場合は、実行時に依存関係をインストールする必要があります。実行時にパッケージをインストールすると、次のような結果が生じる可能性があります。

依存関係の解決、ダウンロード、インストールにより、ワーカーの起動時間が長くなります。
パイプラインを実行するには、インターネットへの接続が必要です。
不確実な処理は、依存関係のソフトウェアリリースによって発生します。

これらの問題を回避するには、カスタム Docker コンテナイメージでランタイム環境を指定します。パイプラインの依存関係がプリインストールされたカスタム Docker コンテナイメージを使用することには、次の利点があります。

Dataflow ジョブを起動するたびに、パイプラインランタイム環境に同じ一連の依存関係が存在する状態が確保されます。
パイプラインのランタイム環境を制御できます。
起動時に時間を要する可能性のある依存関係の解決が行われることを回避します。

カスタムコンテナイメージを使用する場合は、次のガイダンスを考慮してください。

カスタム画像では :latest タグを使用しないでください。ビルドには、日付、バージョンまたは固有識別子をタグとして設定します。このステップにより、必要に応じて既知の動作構成に戻すことができます。
コンテナイメージと互換性のある起動環境を使用します。カスタムコンテナの使用方法の詳細については、コンテナイメージをビルドするをご覧ください。

Python 依存関係のプリインストールの詳細については、Python 依存関係をプリインストールするをご覧ください。

Dataflow テンプレートを使用してリリース環境を制御する

パイプラインに追加の依存関係が必要な場合は、ランタイム環境と起動環境の両方にインストールすることが必要な場合があります。リリース環境では、本番環境バージョンのパイプラインが実行されます。起動環境にはランタイム環境と互換性があることが必要なため、両方の環境で同じバージョンの依存関係を使用します。

コンテナ化された、再現可能な起動環境を作成するには、Dataflow Flex テンプレートを使用します。詳細については、Flex テンプレートをビルドして実行するをご覧ください。Flex テンプレートを使用する場合は、次の要素を考慮してください。

パイプラインをパッケージとして構成する場合は、パッケージをテンプレート Dockerfile にインストールします。Flex テンプレートを構成するには、FLEX_TEMPLATE_PYTHON_SETUP_FILE を指定します。詳細については、必要な Dockerfile 環境変数を設定するをご覧ください。
パイプラインでカスタムコンテナイメージを使用する場合は、テンプレートを起動するときに指定します。詳細については、依存関係にカスタムコンテナを使用するをご覧ください。
Dataflow Flex テンプレートの Docker イメージを構築するには、ベースイメージと同じカスタムコンテナイメージを使用します。詳細については、カスタムコンテナイメージを使用するをご覧ください。

この構成により、リリース環境が再現可能になるとともに、ランタイム環境との互換性が確保されます。

この方法に従った例としては、GitHub にある依存関係のあるパイプラインとカスタムコンテナの Flex テンプレートのチュートリアルをご覧ください。

詳細については、Apache Beam のドキュメントの起動環境とランタイム環境の互換性を確保するとパイプラインが使用する依存関係を制御するをご覧ください。

Go

Apache Beam Go SDK を使用して Dataflow ジョブを実行する場合は、Go モジュールを使用して依存関係を管理します。次のファイルには、パイプラインで使用されるデフォルトのコンパイルとランタイムの依存関係が含まれています。

https://raw.githubusercontent.com/apache/beam/vVERSION_NUMBER/sdks/go.sum

VERSION_NUMBER は、使用している SDK バージョンに置き換えます。

Go パイプラインの依存関係の管理については、Go のドキュメントの依存関係の管理をご覧ください。