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

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

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

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

制限事項

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

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

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

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

Portable 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 以降を使用するバッチ パイプラインの場合、Portable Runner はデフォルトで有効になっています。

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

  • Beam SDK バージョン 2.74 以降: enable_portable_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 バージョンでパイプラインが動作しているにもかかわらず、ご使用のパイプラインで Portable Runner が使用されない場合があります。Portable 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 はデフォルトで有効になっています。

Portable Runner を無効にする

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

Java

Portable Runner を無効にするには、Beam SDK のバージョンに対応するテスト値を設定します。

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

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

Python

Apache Beam Python SDK バージョン 2.45.0 以降では Portable Runner を無効にできません。

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

Go

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

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

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

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

Portable Runner のトラブルシューティング

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 多言語パイプラインに関連する一般的なエラーをデバッグするには、多言語パイプラインのヒントのガイドをご覧ください。