はじめに
AWS MWAA (Amazon Managed Workflows for Apache Airflow) にてワークフロー実行中に表題のエラーに遭遇したため、その解決法について簡潔にまとめます。
何が起きたか？
ワークフローを実行するとエラーが起き、タスクのエラーログには以下のように出ていました。
Executor reports task instance <TaskInstance: hoge_dag.hoge_task manual__yyyy-mm-ddTHH:MM:SS+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
どうやらタスクはキューされているのに、実行できていないようです。これだけではエラーの原因がわからずですが、サーバーログの方に原因がきちんと出ていました。
以下ログ
airflow.exceptions.AirflowTaskTimeout: DagBag import timeout for hoge_dag.py after 30.0s.
Please take a look at these docs to improve your DAG import time:
* https://***.apache.org/docs/apache-***/2.2.2/best-practices.html#top-level-python-code
* https://***.apache.org/docs/apache-***/2.2.2/best-practices.html#reducing-dag-complexity, PID: 909
ERROR - Failed to execute task Dag 'hoge_dag' could not be found; either it does not exist or it failed to parse..
DAGをimportしようとしたが、タイムアウトになってしまっているようですね。
原因
直接的な原因は上で記載した通りタイムアウトです。
以下のAirflowのBest Practicesが載っているページのReducing DAG complexityの項を見ると、タスクの数や分岐など、DAGの構造が複雑になるほどパフォーマンスに影響があることが分かります。
今回の僕のケースでは、分岐を多く含むDAGとして構築していたため、DAGをimportする時間が伸びてしまったと考えられました。
どう解決したか？
原因のところで述べたように、DAGの複雑性を低くするかタイムアウト時間を伸ばすことでエラーは解決します。
今回はタイムアウト時間を伸ばす方法をとります。
エラーログにも出ているように、デフォルトのタイムアウト設定値は30秒なので、これを伸ばしてやります。
タイムアウトの設定はcore.dagbag_import_timeoutの変数で設定されています。
https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#dagbag-import-timeout
MWAAの場合は、airflow.cfgが公開されておらず直接編集できないので、MWAAのリソースを作成する際に環境変数としてセットしてやる必要があります。
僕はCloud Formationにて作成していたため、template.yamlに以下のように追記しました。
MwaaEnvironment:
Type: AWS::MWAA::Environment
Properties:
・
・
AirflowConfigurationOptions:
# dagの読み込みタイムアウト
core.dagbag_import_timeout: 60