Airflow トリガラーに関する問題のトラブルシューティング

Cloud Composer 1 | Cloud Composer 2

このページでは、Airflow トリガーの一般的な問題に関する情報とトラブルシューティングの手順について説明します。

トリガーでのオペレーションのブロック

非同期タスクが triggerer でブロックされる場合があります。ほとんどの場合、問題は triggerer リソースが不十分であるか、カスタム非同期演算子コードに問題があることが原因です。

triggerer ログには、triggerer のパフォーマンスの低下の根本原因を特定できる警告メッセージが表示されます。注意すべき警告が 2 つあります。

  1. 非同期スレッドがブロックされた

    Triggerer's async thread was blocked for 1.2 seconds, likely due to the highly utilized environment.

    この警告は、大量の非同期タスクが原因で発生するパフォーマンスの問題を示しています。

    解決策: この問題に対処するには、triggerer に追加のリソースを割り当てて、同時に実行される遅延タスクの数を減らすか、環境内の triggerer の数を増やします。triggerer が遅延可能なタスクを処理する場合でも、各タスクの開始と完了はワーカーが担います。triggerer の数を調整する場合は、ワーカー インスタンスの数のスケーリングも検討してください。

  2. 特定のタスクで非同期スレッドがブロックされました。

    WARNING - Executing <Task finished coro=<TriggerRunner.run_trigger() done, defined at /opt/***/***/jobs/my-custom-code.py:609> result=None> took 0.401 second

    この警告は、Cloud Composer によって実行される特定のオペレータ コードを示しています。トリガーは、設計上、バックグラウンドでオペレーションを実行するために asyncio ライブラリに依存する必要があります。トリガーのカスタム実装が、asyncio 契約に正しく準拠していない可能性があります（たとえば、Python コードで await キーワードと async キーワードが正しく使用されていない場合など）。

    解決策: 警告に表示されたコードを調べて、非同期オペレーションが正しく実装されているかどうかを確認します。

トリガーが多すぎる

遅延タスクの数は task_count 指標に表示されます。この指標は、環境の Monitoring ダッシュボードにも表示されます。各トリガーは、メモリを消費する外部リソースへの接続など、いくつかのリソースを作成します。

Monitoring ダッシュボードに表示される遅延タスク
図 1。Monitoring ダッシュボードに表示される遅延タスク（クリックして拡大）

メモリと CPU 消費量のグラフは、ハートビートが欠落していることにより liveness プローブが失敗するためにリソース不足で再起動が発生していることを示しています。

リソースが不十分なためにトリガーが再起動する
図 2.リソースが不十分なためにトリガーが再起動する（クリックして拡大）

解決策: この問題に対処するには、triggerer に追加のリソースを割り当てて、同時に実行される遅延タスクの数を減らすか、環境内の triggerer の数を増やします。

コールバック実行中に Airflow ワーカーがクラッシュする

トリガーの実行が完了すると、コントロールが Airflow ワーカーに戻り、実行スロットを使用してコールバック メソッドが実行されます。このフェーズは Celery Executor によって制御されるため、対応する構成とリソースの上限（parallelism や worker_concurrency など）が適用されます。

Airflow ワーカーでコールバック メソッドが失敗した場合、ワーカーが失敗した場、メソッドを実行するワーカーが再起動した場合、タスクは FAILED とマークされます。この場合、コールバック メソッドだけでなく、タスク全体が再実行されます。

トリガー内の無限ループ

カスタム トリガー演算子は、メインの triggerer ループを完全にブロックするように実装できます。これにより、無効なトリガーが一度に 1 つだけ実行されます。この場合、問題のあるトリガーが終了すると、警告がトリガーログに生成されます。

トリガークラスが見つからない

DAG フォルダは Airflow トリガーと同期されないため、トリガーの実行時にインラインのトリガーコードがありません。このエラーは、失敗したタスクのログに生成されます。

ImportError: Module "PACKAGE_NAME" does not define a "CLASS_NAME" attribute/
class

解決策: PyPI から不足しているコードをインポートします。

Airflow UI のトリガーに関する警告メッセージ

トリガーが無効になると、Airflow UI に次の警告メッセージが表示されることがあります。

The triggerer does not appear to be running. Last heartbeat was received
4 hours ago. Triggers will not run, and any deferred operator will remain
deferred until it times out or fails.

不完全なトリガーが Airflow データベースに残っているため、Airflow はこのメッセージを表示できます。このメッセージは通常、環境内ですべてのトリガーが完了する前にトリガーが無効になったことを意味します。

環境で実行されているすべてのトリガーを表示するには、Airflow UI で [閲覧] > [トリガー] ページで確認します（Admin ロールが必要）。

解決策:

トリガーが無効になってもタスクが遅延状態のままである

トリガーを無効にすると、すでに遅延状態にあるタスクはタイムアウトに達するまでこの状態のままになります。Airflow と DAG の構成によっては、このタイムアウトが無限になる場合があります。

次のいずれかの方法を使用します。

  • タスクを手動で失敗としてマークします。
  • タスクを完了するには、トリガーを有効にします。

環境で遅延演算子やタスクが実行されず、遅延タスクがすべて完了している場合にのみトリガーを無効にすることをおすすめします。

次のステップ