Dataflow ポータブル ランナーを使用する

Dataflow を使用してパイプラインを実行すると、Dataflow ランナーはパイプライン コードと依存関係を Cloud Storage バケットにアップロードし、Dataflow ジョブを作成します。この Dataflow ジョブは、Google Cloudのマネージド リソースでパイプラインを実行します。

  • Apache Beam Java SDK バージョン 2.54.0 以降を使用するバッチ パイプラインの場合、ポータブル Runner がデフォルトで有効になっています。
  • Apache Beam Java SDK を使用するパイプラインで、多言語パイプラインの実行、カスタム コンテナの使用、Spanner または Bigtable の変更ストリーム パイプラインの使用を行うには、ポータブル ランナーが必要です。大規模な Java ストリーミング パイプラインには、Streaming Java（デフォルト）Runner を使用します。
  • Apache Beam Python SDK バージョン 2.21.0 以降を使用するパイプラインの場合、Portable Runner がデフォルトで有効になっています。Apache Beam Python SDK バージョン 2.45.0 以降を使用するパイプラインの場合、Dataflow ポータブル ランナーが使用可能な唯一の Dataflow ランナーです。
  • Apache Beam SDK for Go の場合、利用可能な Dataflow ランナーは Portable Runner だけです。

ポータブル ランナーはサービスベースのアーキテクチャを使用しており、多数のパイプラインにメリットがあります。

制限事項

Dataflow Portable Runner には、次の要件と制限事項があります。

  • ポータブル専用の機能: 次の機能は、ポータブル ランナーでのみサポートされています。
    • Java のマネージド変換と RunInference。
    • カスタム コンテナ。
    • ARM ベースのワーカー VM。
    • Splittable DoFn。Non-Portable ランナーのサポートが検討されています。
  • Dataflow Portable Runner では、ストリーミング ジョブに Streaming Engine が必要です。
  • Dataflow Portable Runner にはストリーミング ジョブに Streaming Engine が必要なため、Dataflow Portable Runner を必要とする Apache Beam 変換では、ストリーミング ジョブにも Streaming Engine を使用する必要があります。たとえば、Apache Beam SDK for Python の Pub/Sub Lite I/O コネクタは、Dataflow Portable Runner を必要とする言語横断的な変換です。この変換を使用するジョブまたはテンプレートで Streaming Engine を無効にしようとすると、ジョブは失敗します。
  • Apache Beam Java SDK を使用するストリーミング パイプラインの場合、Portable Runner では MapState クラスと SetState クラスはサポートされていません。Java パイプラインで MapState クラスと SetState クラスを使用するには、Streaming Engine を有効にし、ポータブル Runner を無効にして、Apache Beam SDK バージョン 2.58.0 以降を使用します。
  • Apache Beam Java SDK を使用するバッチ パイプラインとストリーミング パイプラインの場合、AfterSynchronizedProcessingTime クラスはサポートされていません。
  • 多くの場合、ポータブル ランナーは非ポータブル ランナーよりもスケーリングに優れていますが、固定シャーディングではメモリ使用量が多くなる可能性があります。
  • Dataflow の従来のテンプレートは、ビルドされたバージョンとは異なるバージョンの Dataflow ランナーで実行することはできません。つまり、Google 提供の従来のテンプレートではポータブル Runner を有効にできません。カスタム テンプレートでポータブル ランナーを有効にするには、テンプレートをビルドするときに適切なフラグを設定します。このフラグは、使用している SDK バージョンによって異なります。
    • Beam SDK バージョン 2.74 以降: --experiments=enable_portable_runner。
    • Beam SDK バージョン 2.73 以前: --experiments=use_runner_v2。

  • 既知の自動スケーリングの問題により、ステートフル処理を必要とするバッチ Java パイプラインでは、ポータブル Runner はデフォルトで無効になっています。これらのパイプラインでポータブル Runner を有効にすることはできますが（ポータブル Runner を有効にするを参照）、パイプラインのパフォーマンスが大幅に低下するおそれがあります。

  • 一部のパイプラインでは、ポータブル ランナーを使用すると整合性エラーの頻度が増加する可能性があります。ログファイルに次のエラーが表示されます。「Internal consistency check failed, the output is likely incorrect. Please retry the job」。考えられる対応策は、Join/GrouByKey ステップの後に Reshuffle 変換を追加することです。障害率が許容範囲を超えており、緩和策で問題が解決しない場合は、ポータブル ランナーを無効にすることを試してください。

ポータブル Runner を有効にする

Dataflow Portable Runner を有効にするには、Apache Beam SDK の構成手順に沿って操作します。

Java

Dataflow Portable Runner には、Apache Beam Java SDK バージョン 2.30.0 以降が必要です（バージョン 2.44.0 以降を推奨）。

Apache Beam Java SDK バージョン 2.54.0 以降を使用するバッチ パイプラインの場合、ポータブル Runner がデフォルトで有効になっています。

ポータブル ランナーを有効にするには、Beam SDK のバージョンとパイプライン タイプに対応するテスト値を使用してジョブを実行します。

  • Beam SDK バージョン 2.74 以降: enable_portable_runner（バッチ）または enable_streaming_java_runner（ストリーミング）。
  • Beam SDK バージョン 2.73 以前: use_runner_v2。

詳細については、試験運用版のパイプライン オプションを設定するをご覧ください。

Python

Apache Beam Python SDK バージョン 2.21.0 以降を使用するパイプラインの場合、Portable Runner がデフォルトで有効になっています。

Dataflow Portable Runner は、Apache Beam Python SDK バージョン 2.20.0 以前ではサポートされていません。

サポートされている SDK バージョンでパイプラインが動作しているにもかかわらず、ご使用のパイプラインでポータブル ランナーが使用されない場合があります。ポータブル Runner でジョブを実行するには、Beam SDK のバージョンに対応するテスト値を設定します。

  • Beam SDK バージョン 2.74 以降: enable_portable_runner。
  • Beam SDK バージョン 2.73 以前: use_runner_v2。

詳細については、試験運用版のパイプライン オプションを設定するをご覧ください。

Go

Dataflow Portable Runner は、Apache Beam SDK for Go で使用できる唯一の Dataflow ランナーです。Portable Runner はデフォルトで有効になっています。

ポータブル ランナーを無効にする

Dataflow Portable Runner を無効にするには、Apache Beam SDK の構成手順に沿って操作します。

Java

ポータブル ランナーを無効にするには、Beam SDK のバージョンに対応するテスト値を設定します。

  • Beam SDK バージョン 2.74 以降: disable_portable_runner。
  • Beam SDK バージョン 2.73 以前: disable_runner_v2。

これにより、ジョブのデフォルトは Non-Portable Runner（ストリーミング用の Streaming Java Runner）になります。詳細については、試験運用版のパイプライン オプションを設定するをご覧ください。

Python

Apache Beam Python SDK バージョン 2.45.0 以降では、ポータブル ランナーを無効にできません。

以前のバージョンの Python SDK で、ジョブが auto_runner_v2 テストを使用しているものとして識別された場合は、disable_runner_v2 テストを設定することでポータブル Runner を無効にできます。詳細については、試験運用版のパイプライン オプションを設定するをご覧ください。

Go

Go で Dataflow Portable Runner を無効にすることはできません。Portable Runner は、Apache Beam SDK for Go で使用できる唯一の Dataflow ランナーです。

ジョブをモニタリングする

モニタリング インターフェースを使用して、メモリ使用率、CPU 使用率などの Dataflow ジョブの指標を確認します。

ワーカー VM のログは、ログ エクスプローラまたは Dataflow モニタリング インターフェースで利用できます。ワーカー VM のログには、ランナー ハーネス プロセスのログと SDK プロセスのログが含まれます。VM ログを使用して、ジョブのトラブルシューティングを行うことができます。

ポータブル ランナーのトラブルシューティング

Dataflow Portable Runner を使用してジョブのトラブルシューティングを行う場合は、標準のパイプラインのトラブルシューティング手順に沿って行います。次のリストは、Dataflow Portable Runner の仕組みに関する追加情報です。

  • Dataflow Portable Runner を使用するジョブは、ワーカー VM で 2 種類のプロセス（SDK プロセスとランナー ハーネス プロセス）を実行します。パイプラインと VM のタイプによっては、1 つまたは複数の SDK プロセスが存在する場合がありますが、VM ごとに存在するランナー ハーネス プロセスは 1 つのみです。
  • SDK プロセスはユーザーコードやその他の言語固有の関数を実行しますが、それ以外はすべてランナー ハーネス プロセスによって管理されます。
  • ランナー ハーネス プロセスは、Dataflow からの作業のリクエストを開始する前に、すべての SDK プロセスの接続を待機します。
  • SDK プロセスの起動時にワーカー VM が依存関係をダウンロードしてインストールすると、ジョブの遅延が発生することがあります。ライブラリの起動時やインストール時など、SDK プロセス中に問題が発生すると、ワーカーはそのステータスを異常として報告します。起動時間が長くなる場合は、プロジェクトで Cloud Build API を有効にして、次のパラメータを指定してパイプラインを送信します: --prebuild_sdk_container_engine=cloud_build
  • Dataflow Portable Runner はチェックポインティングを使用するため、各ワーカーは変更のバッファリング中に最大 5 秒間待機してから、その変更を送信して処理を続行する場合があります。その結果、約 6 秒のレイテンシが予想されます。
  • ユーザーコードの問題を診断するには、SDK プロセスのワーカーログを調べます。ランナー ハーネスのログにエラーが見つかった場合は、サポートに連絡してバグを報告してください。
  • Dataflow 多言語パイプラインに関連する一般的なエラーをデバッグするには、多言語パイプラインのヒントのガイドをご覧ください。