Cloud Composer(Aiflow)を使う時に実施した方が良い設定をまとめました

# Cloud Composer(Aiflow)を使う時に実施した方が良い設定をまとめました
## はじめに
こんにちは、京セラコミュニケーションシステム　宮坂(@kccs_katsuhiro_miyasaka)です。

皆様、Cloud Composerはご存知でしょうか。
Cloud Composer(Apache Airflow)はワークフローをオーケストレーションします。
画面も見慣れると、エラーも参照でき、詳細なログも確認でき便利なツールです。
しかし、少々不便に感じる事もあります。
そんな人向けに便利な設定等々をお伝えできればと思います。
※今回はAirflow V1を前提と致します。

:::note info
執筆時のComposer(Airflow V1)の最新バージョン
Composer 1.20.1
Airflow  1.10.15
を前提とします。
:::

## Airflowのヘッダーカラーを変える
Composer作成直後のAirflow画面は、以下の画面になります。
どこのプロジェクトでも同様の画面になってしまうので、環境の見間違いが発生しないか不安になります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/973d9469-bd5a-dbd5-6594-fd2f32c104af.png)

そんな方にオススメなのが以下の設定
帯の色を変えて、プロジェクト等の違いを可視化できます。
一目見てわかる事により、誤操作防止を期待できます。
[navbar_color](https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#navbar-color "navbar_color")

具体的な設定方法は、Airflow構成のオーバーライドを編集することで反映されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/b01605f9-8a11-59c9-9b8d-d9031032a70a.png)

以下の形式で指定し、保存します。

|項目名|値|
| ---- | ---- |
|セクション1|webserver|
|キー1|navbar_color|
|値1|#rrggb形式で色指定|

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/536a6248-2881-5018-e206-8634d9010843.png)

すると、帯の色が変わります。
これでプロジェクト設定による差分に気づきやすくなります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/58e419e5-6374-8b43-3a39-8f17e67540a4.png)

## タスクがゾンビプロセスと判定されてkillされてしまう。
タスクが重く時間がかかる時に、Airflowのメインプロセスに応答を返せないと、タスクが終了させられてしまう事が有ります。
そんな時に以下の値を設定します。

[scheduler_zombie_task_threshold](https://airflow.apache.org/docs/apache-airflow/1.10.14/configurations-ref.html#scheduler-zombie-task-threshold"scheduler_zombie_task_threshold")

以下のGoogle Cloudのトラブルシューティングにもあります。
[ゾンビタスクのトラブルシューティング](https://cloud.google.com/composer/docs/composer-2/troubleshooting-dags?hl=ja#troubleshooting_zombie_tasks"ゾンビタスクのトラブルシューティング")

初期値は300秒となっており、Taskの実行時間に合わせてセットする必要があります。

|項目名|値|
| ---- | ---- |
|セクション1|scheduler|
|キー1|scheduler_zombie_task_threshold|
|値1|設定した時間(秒)|

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/f986d802-00b0-82c2-9e52-bf34102c955d.png)


## AirflowのDBにデータが貯まってしまい、DBのレスポンスエラーになる。
突然ですが、Cloud Composer(Airflow)がどういう構成をしているかご存知でしょうか。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/ea9d4f3c-7b6d-84bc-116f-130e69e1d60f.png)
[パブリック IP 環境のアーキテクチャより](https://cloud.google.com/composer/docs/concepts/architecture?hl=ja#public-ip "パブリック IP 環境のアーキテクチャ")

こういった構成になっています。
この構成でAirflowの実行結果をAirflowのDBに保存します。

しばらくたつと、Airflowの実行履歴が大量に保存されてDBのレスポンスが悪くなります。
すると、AirflowがMySQLとの接続エラーになる事があります。
そんな時のためにメンテナンス用のDAGが用意されていますので、そちらをインストールします。

以下に公開されているので、ダウンロードします。
[Airflow データベースをクリーンアップする](https://cloud.google.com/composer/docs/cleanup-airflow-database?hl=ja
"Airflow データベースをクリーンアップする")

何か所か注意点があります。
58行目付近の値を自分で修正する必要があります。
```python
# airflow-db-cleanup
DAG_ID = os.path.basename(__file__).replace(".pyc", "").replace(".py", "")
START_DATE = airflow.utils.dates.days_ago(1) 
# How often to Run. @daily - Once a day at Midnight (UTC)
SCHEDULE_INTERVAL = "@daily"
# Who is listed as the owner of this DAG in the Airflow Web Server
DAG_OWNER_NAME = "operations"
# List of email address to send email alerts to if this job fails
ALERT_EMAIL_ADDRESSES = []
# Length to retain the log files if not already provided in the conf. If this
# is set to 30, the job will remove those files that arE 30 days old or older.

DEFAULT_MAX_DB_ENTRY_AGE_IN_DAYS = int(
    Variable.get("airflow_db_cleanup__max_db_entry_age_in_days", 30))
```

具体的には
・タスクの開始日時を指定します。
```python
airflow.utils.dates.days_ago
```
・実行間隔を指定します。
```python
SCHEDULE_INTERVAL = "@daily"
```
・何日分のデータを保存しておくか指定します。
```python
DEFAULT_MAX_DB_ENTRY_AGE_IN_DAYS = int(
    Variable.get("airflow_db_cleanup__max_db_entry_age_in_days", 30))
```

修正したら、保存しDAGを配置します。
「DAフォルダを開く」を選びます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/89352a30-6f5f-396b-c894-7bad1aeb07de.png)

バケットが表示されるので、ファイルを配置します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/9b3fb55f-4b70-ac65-491b-e68f1e774fdb.png)

配置後しばらくすると反映されます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/a0ae5925-80da-3770-0875-96639df3d274.png)

後は期限が来ると実行されます。
実行後は履歴情報が削除されて、DAG Runsの値が小さくなります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/61650cb7-db8c-b0bb-20fa-ac26b88b4ebb.png)

## DAGのリトライ設定について
システム運用時に処理が失敗時にリトライをしたいケースが有ります。
何回リトライするのか、事前に決められない時に設定値に持たせると便利です。
具体的には、DAGの共通引き数を定義されると思います。
```python
default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}
```

ここに以下の設定を入れます。
```python
#modelsのimport
from airflow import models
#---中略
default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': int(models.Variable.get('retries')),
    'retry_delay': timedelta(minutes=int(models.Variable.get('retry_delay'))
}
```
そして、adminのValiablesを指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/2e75bf91-20ab-449c-33ad-77a6e3d7582c.png)

そして、キーを作るとリトライ設定を外からセットする事ができます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1525729/f8f5bacd-0be7-a26f-e344-2f8cfb814897.png)

これによって、処理失敗の傾向を見ながら後から、リトライの設定を入れる事ができるようになります。

## 終わりに
Composerを使い始めて、苦悩してきたことを上記のような方法で解決してきました。
上記以外の解決方法や、便利な使い方などありましたら、是非コメント頂けますでしょうか。