遅いジョブや停止したジョブのトラブルシューティング

このページでは、Dataflow のストリーミング ジョブやバッチジョブの速度が遅い場合や停止している場合の一般的な原因とそのトラブルシューティング方法について説明します。

ストリーミング

次の症状がある場合、Dataflow ストリーミング ジョブの実行速度が遅くなっているか、停止している可能性があります。

  • パイプラインがソースからデータを読み取っていない。たとえば、Pub/Sub のバックログが増加しています。
  • パイプラインがシンクにデータを書き込まない。
  • データの更新頻度指標が上昇している。
  • システム レイテンシ指標が増加している。

次のセクションの情報を参照して問題を診断してください。

繰り返し発生するエラーを調査する

ストリーミング ジョブでは、一部のエラーが無期限に再試行されます。この再試行により、パイプラインの進行が妨げられます。繰り返し発生しているエラーを特定するには、ワーカーログで例外を確認します。

異常なワーカーを特定する

ストリーミング ジョブを処理するワーカーが異常な状態である場合、そのジョブは低速であるか、停止しているように見えることがあります。異常なワーカーを特定するには:

ストラグラーを特定する

ストラグラーは、ステージ内の他の作業アイテムと比較して低速な作業アイテムです。ストラグラーの特定と修正については、ストリーミング ジョブのストラグラーのトラブルシューティングをご覧ください。

不十分な並列処理のトラブルシューティング

スケーラビリティと効率性を高めるため、Dataflow は複数のワーカーでパイプラインのステージを並行して実行します。Dataflow の並列処理の最小単位はキーです。融合された各ステージの受信メッセージはキーに関連付けられます。キーは、次のいずれかの方法で定義されます。

  • Pub/Sub パーティションなどのソースのプロパティによって暗黙的に定義される。
  • パイプラインの集約ロジック（GroupByKey など）によって明示的に定義される。

パイプラインで特定のステージに十分なキーがない場合、並列処理が制限されます。そのステージがボトルネックになる可能性があります。

並列処理が少ないステージを特定する

パイプラインの速度低下の原因が並列処理の不足であるかどうかを確認するには、CPU 使用率の指標を使用します。CPU の負荷が低く、ワーカーに均等に分散されている場合、ジョブの並列処理が不十分な可能性があります。ジョブで Streaming Engine を使用している場合、ステージの並列処理が不十分かどうかを調べるには、[ジョブの指標] タブで並列処理の指標を確認します。この問題には次のように対処してください。

  • Google Cloud コンソールの [ジョブ情報] ページで、[自動スケーリング] タブを使用して、ジョブのスケールアップに問題があるかどうかを確認します。自動スケーリングが問題である場合は、Dataflow の自動スケーリングのトラブルシューティングをご覧ください。
  • ジョブグラフを使用して、ステージ内のステップを確認します。ステージがソースからの読み取りまたはシンクへの書き込みを行っている場合は、ソースまたはシンクのサービスのドキュメントをご覧ください。ドキュメントを参照して、そのサービスが十分なスケーラビリティを備えているかどうかを判断してください。

ホットキーを確認する

タスクがワーカー間で均等に分散されておらず、ワーカーの利用率が非常に低い場合は、パイプラインにホットキーが存在する可能性があります。ホットキーは、他のキーと比較して処理する要素が非常に多いキーです。この問題を解決するには、次の操作を 1 つ以上行います。

  • データを再入力します。新しい Key-Value ペアを出力するには、ParDo 変換を適用します。詳細については、Apache Beam のドキュメントで Java ParDo 変換ページまたは Python ParDo 変換ページをご覧ください。
  • Combine 変換で .withFanout を使用します。詳しくは、Java SDK の Combine.PerKey クラスまたは Python SDK の with_hot_key_fanout オペレーションをご覧ください
  • 制限のない PCollections を大量に処理する Java パイプラインがある場合は、次のことを行うことをおすすめします。
    • Combine.Globally ではなく Combine.Globally.withFanout を使用します。
    • Count.PerKey ではなく Combine.PerKey.withHotKeyFanout を使用します。

割り当ての不足を確認する

ソースとシンクに十分な割り当てがあることを確認します。たとえば、パイプラインが Pub/Sub または BigQuery から入力を読み取る場合、Google Cloud プロジェクトの割り当てが不足している可能性があります。これらのサービスの割り当て上限の詳細については、Pub/Sub 割り当てまたは BigQuery の割り当てをご覧ください。

ジョブで多数の 429 (Rate Limit Exceeded) エラーが発生している場合は、割り当てが不足している可能性があります。エラーを確認するには、次の操作を行います。

  1. Google Cloud コンソールに移動します。
  2. ナビゲーション パネルで、[API とサービス] をクリックします。
  3. メニューで [ライブラリ] をクリックします。
  4. 検索ボックスを使用して Pub/Sub を検索します。
  5. [Cloud Pub/Sub API] をクリックします。
  6. [管理] をクリックします。
  7. [レスポンス コード別のトラフィック] グラフで、(4xx) クライアント エラーコードを探します。

Metrics Explorer で割り当ての使用状況を確認することもできます。パイプラインで BigQuery ソースまたはシンクを使用している場合、割り当ての問題のトラブルシューティングを行うには、BigQuery Storage API 指標を使用します。たとえば、次の手順で BigQuery の同時接続数を示すグラフを作成します。

  1. Google Cloud コンソールで [Monitoring] を選択します。

    [Monitoring] に移動

  2. ナビゲーション パネルで、[Metrics Explorer] を選択します。

  3. [Select a metric] ペインの [指標] で、[BigQuery Project] > [Write] > [concurrent connection count] を選択します。

Pub/Sub 指標の表示手順については、「Cloud Monitoring で Pub/Sub をモニタリングする」の割り当て使用状況のモニタリングをご覧ください。BigQuery 指標の表示手順については、「ダッシュボード、グラフ、アラートを作成する」の割り当ての使用状況と上限を表示するをご覧ください。

バッチ

バッチジョブが遅いか停止している場合は、[実行の詳細] タブでジョブの詳細情報を確認し、ボトルネックの原因となっているステージやワーカーを特定します。

ストラグラーを特定する

ストラグラーは、ステージ内の他の作業アイテムと比較して低速な作業アイテムです。ストラグラーの特定と修正については、バッチジョブのストラグラーのトラブルシューティングをご覧ください。

遅延しているステージまたは停滞しているステージを特定する

遅延しているステージまたは停滞しているステージを特定するには、[ステージの進捗状況] ビューを使用します。バーが長いほど、ステージに時間がかかっていることを示します。このビューを使用して、パイプラインで最も遅いステージを特定します。

ボトルネックのステージが見つかったら、次の操作を行います。

  • ステージ内で遅延しているワーカーを特定します。
  • 遅延しているワーカーがない場合は、[ステージ情報] パネルで最も遅いステップを特定します。この情報を使用して、ユーザーコードの最適化が必要な場所を特定します。
  • 並列処理のボトルネックを見つけるには、Dataflow モニタリング指標を使用します。

遅延しているワーカーを特定する

特定のステージで遅延しているワーカーを特定するには、[ワーカーの進捗状況] ビューを使用します。このビューでは、ステージが終了するまですべてのワーカーが作業中であるか、特定のワーカーが遅延タスクで行き詰っているかどうかを確認できます。遅延しているワーカーが見つかった場合は、次の操作を行います。

デバッグツール

パイプラインが遅い場合や停滞している場合は、次のツールを使用して問題を診断できます。

  • インシデントと関連付けてボトルネックを特定するには、Dataflow 用 Cloud Monitoring を使用します。
  • パイプラインのパフォーマンスをモニタリングするには、Cloud Profiler を使用します。
  • 変換の中には、他の変換よりも大容量のパイプラインに適しているものがあります。ログ メッセージを使用すると、バッチ パイプラインまたはストリーミング パイプラインで停止したユーザー変換を特定できます。
  • 停止したジョブの詳細については、Dataflow ジョブの指標をご覧ください。有用な指標は次のとおりです。
    • バックログのバイト数の指標（backlog_bytes）は、処理されていない入力量をステージごとにバイト単位で測定します。この指標を使用して、スループットのない融合されたステップを探します。同様に、バックログ要素の指標（backlog_elements）は、ステージ内の未処理の入力要素の数を測定します。
    • 並列処理キー（processing_parallelism_keys）指標は、直近 5 分間のパイプラインの特定のステージにおける並列処理キーの数を測定します。この指標を使用して、次のように調査します。
      • 問題を特定のステージに絞り込み、A hot key ... was detected などのホットキーの警告を確認します。
      • 不十分な並列処理によるスループットのボトルネックを見つけます。これらのボトルネックにより、パイプラインの遅延や停止が発生する可能性があります。
    • システムラグ指標（system_lag）とステージごとのシステムラグ指標（per_stage_system_lag）は、データが処理された最大時間または処理を待機している最大時間を測定します。これらの指標を使用して、データソースから非効率的なステージとボトルネックを特定します。

Dataflow モニタリング ウェブ インターフェースに含まれていない追加の指標については、Google Cloud 指標で Dataflow 指標の完全なリストをご覧ください。