忙しい人のためのAirflowゾンビ撃退法

Cloud Composerのチューニングの話です。
Cloud Composerにはモニタリングという非常にありがたいものが存在します。そこでCPU、メモリが限界まで達していないか確認し、必要に応じてワークロードの構成を変更しましょう。
core_concurrency （ = タスクの同時実行数）を調整してもいいかもしれません。

はじめに

こんにちは！Intimate Merger(インティメート・マージャー)のOharaです。

弊社ではワークフロー管理ツールであるAirflowを活用しています。

AirflowはPythonで記述され、豊富なプラグインや柔軟なカスタマイズ性があります。DAG（Directed Acyclic Graph）と呼ばれるタスクの依存関係を定義するグラフ構造を使用して、複雑なワークフローを効果的に管理できます。
弊社ではGoogle CloudのCloud Composerを用いて、Airflowをセットアップ&運用しています。

12日目にAirflowの関連記事がありますので、そちらもぜひ！
→ Airflowスケジューリングマスターへの最初の一歩

本記事では、そんなAirflowを楽しく使っていた時に突如襲いかかってきたゾンビたちに、どう立ち向かい駆逐していったかを、後世に残していきたいと思います。

前提

当時のversionは以下です。

composer-2.0.24-airflow-2.2.5

また、ワークロードの構成は次の通りでした。

	vCPU	メモリ	ストレージ	数
スケジューラ	1 vCPU	1.875 GB	1 GB	1
ワーカー	0.5 vCPU	1.875 GB	1 GB	1~3
ウェブサーバー	2 vCPU	3 GB	1 GB	-

Cloud Composer、Airflowのバージョンアップに伴い、Airflow1.x系で動かしていたDagsをAirflow2.x系に移行している最中でした。

ゾンビ、第一波襲来

それは深夜2時のこと…悲劇は突然訪れる…

ギャー😱

複数のタスクが Detected <TaskInstance: hogehoge> as zombie というエラー文とともに異常終了するという現象が起こりました。

原因調査

調査の結果、このzombieは、特定のDAGを追加してから急に発生し始めていることがわかりました。

その後、開発環境でDAGを実行し、どのくらいの負荷がかかるかを調べた結果、複数のtaskを同時に実行すると当然のことながらtaskの分だけメモリやCPUを大量消費していました。

このDAG自体は、最大24taskを同時に実行するような構成となっています。したがって、対応策として以下の2つが考えられます。

メモリやCPUを上げて、24taskの同時実行に耐えうるスペックにする
同時に実行できるtaskの数を減らす

今回はtaskの数を調整するために celery.worker_concurrency に着目しました。

celery.worker_concurrency は、1つのワーカーがタスクキューから選択できるタスクの数を調整できます。

対処

celery.worker_concurrency はAirflow2.2.3以降では、「 32 、12 * worker_CPU 、8 * worker_memory の最小値」（Cloud Composer公式ドキュメント参照）と定義されているため、当時の構成ではdefault値は 6 となります。

（Airflow公式ドキュメントには 16 とあるのですが、Cloud Composerが適切に設定してくれているようです。実際には airflow.cfg にAirflow構成の詳細があり、こちらを参照するといいでしょう。）

Cloud ComposerでAirflowを立ち上げると自動的に作成されるGoogle Cloud Storageのバケット内に、Airflow構成の詳細が記載された airflow.cfgファイルがあります。

当時対応した際は、参考にしていたこちらの記事に

worker_concurrency = 6-8 * core数 または 6-8 * memory/3.75GB

が推奨値であると記載されていたため、メモリが1.875GBであったことを考慮し worker_concurrency を 4 に設定したところ無事ゾンビ化を抑えることができました。

ただ、DAGが今後増えることが予想されていたので、実際は以下のように変更しました。

【ワーカー】

vCPU 0.5 → 1
メモリ 1.875 → 3.75

【Airflow構成のオーバーライド】

celery.worker_concurrency default (6) → 8

推奨値が書いてあった記事は2020年のものなので、Cloud Composer1, Airflow1.x系に当てはまる内容かもしれません。今回はうまくいきましたが、あくまで参考程度にとどめておいた方が良いかと思います。

こうして、バランスをうまく調整することで大量発生したゾンビたちを駆逐することに成功しました。しかし、その平穏も長くは続かないのです…。

DAGのimportがされない？

ゾンビが消えてから数ヶ月経った後、今度はDAGのimportがタイムアウトするという現象に悩まされるようになりました。

Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/site-packages/pkg_resources/__init__.py", line 2086, in dist_factory
    lower.endswith('.dist-info') and
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/utils/timeout.py", line 37, in handle_timeout
    raise AirflowTaskTimeout(self.error_message)
airflow.exceptions.AirflowTaskTimeout: DagBag import timeout for /home/airflow/gcs/hogehoge.py after 30.0s.
Please take a look at these docs to improve your DAG import time:
* https://airflow.apache.org/docs/apache-airflow/2.2.5/best-practices.html#top-level-python-code
* https://airflow.apache.org/docs/apache-airflow/2.2.5/best-practices.html#reducing-dag-complexity, PID: 3849