Cloud Composer Tips

皆さんこんにちは。@best_not_bestです。

今年は[Cloud Composer](https://cloud.google.com/composer)を利用することが多かったため、そこで得たTipsをまとめたいと思います。
なお、以下記載のコードはPython 3での動作を想定しております。

## サポート対象のPythonバージョン

2.7.15と3.6.6となります。
https://cloud.google.com/composer/docs/concepts/python-version

## タスク失敗時のロギング/Monitoring検知

タスクが失敗した時のロギングへのログの落ち方がなかなか分かりにくいです。
`AirflowException`を`raise`するタスクを用意し、これを各タスクの後に`one_failed`で実行させ、この時のメッセージを検知するのがシンプルかなと思います。

### サンプルプログラム

```py3:sample_1.py
#!/usr/bin/env python
# -*- coding: UTF-8 -*-

"""sample."""

from airflow import AirflowException
from airflow.operators.python_operator import PythonOperator

def output_error():
    """output error."""
    raise AirflowException('task failed.')

# （省略）

with models.DAG(
    dag_id='sample_001',
    # （省略）
) as dag:
    task_1 = # （省略）
    task_2 = # （省略）
    error_task = PythonOperator(
        task_id='error_task',
        python_callable=output_error,
        retries=0,
        trigger_rule='one_failed',
        dag=dag,
    )

    task_1 >> error_task
    task_2 >> error_task
    task_1 >> task_2
```

### ロギングでの指標

以下で指標を作成します。

```
resource.type="cloud_composer_environment"
logName=projects/<プロジェクトID>/logs/airflow-worker
labels.workflow:sample_001
("task failed.")
```

![スクリーンショット 2020-12-17 2.20.28.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/17452/f8626c28-cac8-131e-9f3c-694abcc42805.jpeg)

### Monitoringでのアラートポリシー

上記指標をターゲットとしてアラートポリシーを作成します。

![スクリーンショット 2020-12-17 2.38.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/17452/f1d1125f-0abc-f4fd-a93f-216c97a88a0c.png)
![スクリーンショット 2020-12-17 2.38.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/17452/d1a50ba3-f206-fc70-ba8f-4323a522e2ac.png)

## エラーログ

ロギングでは少々分かりにくいので、Composerバケット配下のログが分かりやすいかと思います。

![スクリーンショット 2020-12-17 2.46.01.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/17452/6cb2e94f-e0c3-7dab-d182-2759e27f3e03.jpeg)

## タスクの考え方

`PythonOperator`である程度自由に処理を記述できますが、用意されているOperatorを極力使った方が処理速度が速いです。
https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/index.html
https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/contrib/operators/index.html

例えばBigQuery内のデータをCloud StorageにCSVファイル出力する処理であれば、

1. `PythonOperator`でBigQueryのデータをクエリで取得
2. `PythonOperator`で取得したデータをCSVファイルにしてローカルに保存
3. `PythonOperator`でCSVファイルをCloud Storageにコピー

ではなく

1. `BigQueryOperator`でクエリ実行結果をBigQuery上にテーブル保存
2. `BigQueryToCloudStorageOperator`でテーブル内のデータをCloud Storageに出力

の方が、BigQueryやCloud Storage側で処理が行われるため実行時間が速くなります。
（前者はデータ数によっては処理が終わらないことも・・・。）

## VPC Service Controls

[境界内に置くことは可能](https://cloud.google.com/composer/docs/configuring-vpc-sc)ですが、Pythonパッケージのインストールや外部システムとの連携の際の設定が複雑になる、等の理由から個人的には利用していません。
BigQueryやCloud Storageだけでも境界内に置きたいという場合は、

* BigQueryやCloud Storageを扱うプロジェクトA（境界内）
* Cloud Composerを扱うプロジェクトB（境界外）

のようにプロジェクトを分け、Cloud Composerを実行しているサービスアカウントからのアクセスをプロジェクトAで許可するようにすると後々楽かと思います。

![スクリーンショット 2020-12-17 3.34.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/17452/cf662328-54e1-5f4d-6956-db8da90e27c3.png)

また余談ですが、セキュリティ向上のためとはいえFWルールで下りを全遮断してしまうと、スケジューラ等が動かなくなるのでご注意ください。（[Tenant Project](https://cloud.google.com/composer/docs/concepts/architecture)と通信しているため？）
可能な限り下りは開けたままの方が良いかと思います。

## ヘルスチェック

以下ご参考ください。
https://cloud.google.com/composer/docs/tutorials/health-check

## まとめ

* タスク失敗検知は検知用のタスクを用意する
* 用意されているOperatorを極力使いましょう
* 過度なVPC Service ControlsやFWルールでのアクセス制限は、保守運用を複雑にする場合もある

以上、ご参考になれば幸いです！