Cloud Composer 簡易フレームワーク作ってみた

# 1. 機能
**BigQueryExecuteQueryOperator**の直列実行タスクを動的に生成する機能。
このままでもデータのバックフィルなどでの利用できますが、他の用途のタスクを追加したいときなどは下記のリポジトリを参考に作ってみてください。

[Github - sample-cloud-composer](https://github.com/kirakirapq/sample-cloud-composer)

# 2. ディレクトリ構成

```shell
.
├── dags - ワークフローを設置する
│   ├── project_1
│   │   └── dag1.py
│   ├── dag1.py
│   └── dag2.py
├── data
│   └── key
│   │    └── service-account-key.json - サービスアカウントキー
│   └── slack
│   │    └── failure.text - 失敗通知テンプレート
│   │    └── success.text - 成功通知テンプレート
│   └── sql - SQL置き場
|   │     └── project_1
│   │        └── test1.sql
│   │        └── test2.sql
│   └── test
|         └── project_1
│                └── dag1.py - dag1ワークフローをテスト
│                └── dag2.py - dag2ワークフローをテスト
└── plugins
    └── modules - モジュール置き場
└── variables
    └── base.json -共通設定定義
    └── project_1.json - project_1共通設定定義(project_id, datasetなど)

```

|ディレクトリ|用途|備考|
|:--|:--|:--|
|./data|データ関連を置く|testやslack(テンプレート）やsqlなどを置いてます|
|./plugins|モジュール置き場||
|./valiables | Airflowの共通設定情報を定義|プロジェクトごとの共通設定: {project_name}.json|
|./{project_1} | project_1のDAGを定義|プロジェクトごとのdagファイル置き場|


# 3. セットアップ
## 3-1. 準備
1. `dags`直下に`project_nameディレクトリ`を作成してDAGを設置してください
2. プロジェクトで使用するsqlを「./data/sqls/{project_name}」配下へ置きます
3. テストワークフローを「./data/test/{project_name}」配下へ置きます
4. 共通設定がある場合は「./valiables/{env}/{project_name}.json」を作成します

## 3-2. Cloud Composerの環境構築
まずはGCPへCloud Composerをセットアップします
Cloud Composerのセットアップは25分ほどがかかりますが、下記コマンドでセットアップは完了します
セットアップが完了するとGKEインスタンスにairflowサーバーが作成されます
また、ワークフローの保存先としてCloud Composer用のCloud Storageも作成されます

```shell
export ENVIRONMENT_NAME=your-composer-name
export ENVIRONMENT_LOCATION=asia-northeast1
export PROJECT_ID=your-project-id
export DEPLOY_ENV=develop
 
# airflow.cfgの設定を上書き
export worker_concurrency="celery-worker_concurrency=2"
export worker_autoscale="celery-worker_autoscale=2,1"
export max_active_runs_per_dag="core-max_active_runs_per_dag=128"
export dag_concurrency="core-dag_concurrency=128"
export parallelism="core-parallelism=128"
export max_threads="scheduler-max_threads=10"
 
gcloud composer environments create ${ENVIRONMENT_NAME} \
      --location ${ENVIRONMENT_LOCATION} \
      --zone asia-northeast1-b \
      --airflow-configs=^:^${worker_concurrency}:${worker_autoscale}:${max_active_runs_per_dag}:${dag_concurrency}:${parallelism}:${max_threads} \
      --machine-type n1-standard-1 \
      --disk-size 100 \
      --env-variables DEPLOY_ENV=${DEPLOY_ENV},PROJECT=${PROJECT_ID} \
      --node-count 3 \
      --python-version 3
```
## 3-3. 設定確認
### 3-3-1. ENVIRONMENT CONFIGURATION
環境基本設定が正しく設定されていることを確認する
![スクリーンショット 2021-05-06 15.18.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294552/b778c209-88d0-e54c-ade5-b790a2db6cce.png)
### 3-3-2. AIRFLOW構成のオーバーライド
airflow.cfgが正しく設定されていることを確認する
より詳細な設定値はAirflow UI > Admin > Configurationからも確認できます
![スクリーンショット 2021-05-06 15.17.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294552/43ab6cf8-b24c-79ce-bc9d-2103d00a2e5c.png)
### 3-3-3. 環境変数
環境変数が設定されていることを確認
![スクリーンショット 2021-05-06 15.17.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294552/580cf162-d96d-87ac-ba4f-030368a2d10b.png)

# 4. Connectionを作成
## 4-1. Connection定義
|Conn Id|Conn Type|Project Id|Key Path|
|:--|:--|:--|:--|
|your-conn-id|google_cloud_platform|your-project-id|/home/airflow/gcs/data/keys/conn_id.json

## 4-2. 新しい接続を作成
```shell
export ENVIRONMENT_NAME=your-composer-name
export ENVIRONMENT_LOCATION=asia-northeast1
export PROJECT_ID=your-project-id
export KEY_PATH=/home/airflow/gcs/data/keys/conn_id.json
export CONN_ID=your-conn-id
 
 
gcloud composer environments run ${ENVIRONMENT_NAME} \
 --location ${ENVIRONMENT_LOCATION} connections -- --add \
 --conn_id=${CONN_ID} --conn_type=google_cloud_platform \
 --conn_extra='{"extra\__google\_cloud\_platform\__project": "${PROJECT_ID}",
 "extra\__google\_cloud\_platform\__key_path": "${KEY_PATH}",
 "extra\__google\_cloud\_platform\__scope": "https://www.googleapis.com/auth/cloud-platform"}'
```

## 4-3. 新しい接続の作成/更新画面
* Airflow UI > Admin > Connection > Listの順に遷移して"4-1で作成したconn_id"の編集ボタンをクリックして下記情報を登録

|項目名|値|
|:-|:-|
|Conn Id|設定済|
|Conn Type|設定済|
|Project Id|${PROJECT_ID}|
|Keyfile Path|${KEY_PATH}|
![スクリーンショット 2021-05-06 18.59.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294552/eaed32cc-bde1-2ca8-83e6-248a59e40354.png)
## 4-4. Key Fileを登録
```sh
export ENVIRONMENT_NAME=your-composer-name
export ENVIRONMENT_LOCATION=asia-northeast1
# ~/Documents/key/composer-workflow-test.json : keyファイルを指定
export KEY_FILE_PATH=$(cp ~/Documents/key/your_key.json ./data/keys/conn_id.json | echo ./data/keys)
 
# keyを登録
gcloud composer environments storage data import \
        --environment ${ENVIRONMENT_NAME} \
        --location ${ENVIRONMENT_LOCATION} \
        --source ${KEY_FILE_PATH}
 
# リポジトリからキーファイルを削除
rm ./data/keys/conn_id.json
```

# 5. Variablesを登録
## 5-1. Variables定義一覧
|キー名|値|説明|
|:-|:-|:-|
|module_path|/home/airflow/gcs/plugins/modules/|モジュールパス|
|sql_path|/home/airflow/gcs/data/sqls/|SQLパス|
|slack_subject|composer-workflow-dev - Dag:{0[Dag]}|Slack通知の件名|
|slack_url||slack web hook url|
|slack_channel|tmp-composer-notification|Slack通知先チャンネル名|
|slack_failure_text||/home/airflow/gcs/data/slack/slack_failure_test.txt|タスク失敗時のSlack通知の本文のテンプレート|
|slack_success_text|/home/airflow/gcs/data/slack/slack_success_test.txt|タスク成功時のSlack通知の本文のテンプレート|
|conn_id|your-conn-id|airflow conn id|

## 5-2. Variables登録手順
1. Airflow UI > Admin > Variables の順に遷移
2. 「ファイルを選択」ボタンをクリックして「.variables/${DEPLOY_ENV}/airflow.Variable.json」を選択
3. 「Import Variables」ボタンをクリックして設定を取り込む
![スクリーンショット 2021-05-06 19.00.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/294552/3734fe90-e37e-398f-8fc3-b34a08b7e952.png)

# 6. DAG / Data / Plugin のアップロード
## 6-1. 環境変数

|変数名|値|
|:--:|:--:|
|ENVIRONMENT_NAME|環境名|
|ENVIRONMENT_LOCATION|ロケーション|
|PROJECT_NAME|プロジェクト名|

## 6-2. アップロード

|アップロード場所|説明|
|:--|:--|
|airflow.cfg|Airflowの設定ファイル|
|env_var.json|環境変数|
|bucket_root/dags/{PROJECT_NAME}/|定義済のDAGは各プロジェクトごとにディレクトリを切って設置|
|bucket_root/datas/slack/|slack通知用テンプレート|
|bucket_root/datas/sqls/{PROJECT_NAME}/|SQLは各プロジェクトごとにディレクトリを切って設置|
|bucket_root/datas/test/{PROJECT_NAME}/|テストファイルは各プロジェクトごとにディレクトリを切って設置|
|bucket_root/plugins/modules/|共通モジュール置き場|

### 6-2-1. DAGsをアップロード

```sh
gcloud composer environments storage dags import \
        --environment ${ENVIRONMENT_NAME} \
        --location ${ENVIRONMENT_LOCATION} \
        --source ./dags/${PROJECT_NAME}
```
### 6-2-2. slack通知用テンプレートをアップロード

```sh
gcloud composer environments storage data import \
        --environment ${ENVIRONMENT_NAME} \
        --location ${ENVIRONMENT_LOCATION} \
        --source ./data/slack
```
### 6-2-3. sqlをアップロード

```sh
gcloud composer environments storage data import \
        --environment ${ENVIRONMENT_NAME} \
        --location ${ENVIRONMENT_LOCATION} \
        --source ./data/sqls
```
### 6-2-4. テストをアップロード

```sh
gcloud composer environments storage data import \
        --environment ${ENVIRONMENT_NAME} \
        --location ${ENVIRONMENT_LOCATION} \
        --source ./data/test
```
### 6-2-5. 共通モジュールをアップロード

```sh
gcloud composer environments storage plugins import \
        --environment ${ENVIRONMENT_NAME} \
        --location ${ENVIRONMENT_LOCATION} \
        --source ./plugins/modules
```







# 7. ワークフロー実装
[サンプル-TEST_PRJ/T001.py](https://github.com/kirakirapq/sample-cloud-composer/blob/master/TEST_PRJ/T001.py)

## 7-1. DAGの作成
* DAGを生成します

```sample.py
one_day_ago = datetime.combine(datetime.today() - timedelta(1), datetime.min.time())
original_args = {
        "owner": "airflow",
        "depends_on_past": False,
        "start_date": one_day_ago,
        "email": ["airflow@airflow.com"],
        "email_on_failure": False, # Trueにするとtaskが失敗した際にメールが通知
        "email_on_retry": False, # Trueにするとtaskがリトライが発生した際にメールが通知
        "retries": 1,
        "retry_delay": timedelta(minutes=10)}

schedule_interval = timedelta(days=1)

dag = DAG("TEST_PRJ-T001", default_args=original_args, schedule_interval=schedule_interval)
```

* default_args['end_date'] < excute_date(現在日）にするとワークフローが実行されないので過去〜過去のbackfillを行いたい時は下記コマンドで行う

```
gcloud composer environments run ${ENVIRONMENT_NAME} \
    --location ${ENVIRONMENT_LOCATION} \
    backfill \
    -- -s ${START_DATE} -e ${END_DATE} -B ${DAG_NAME}
```
## 7-2. ワークフローを作成

```sample.py
interval = create_interval("weekly", 1)

destination_dataset_table = "your_project_id.dataset.table${{ params.partition_date }}"

task01 = get_big_query_task_model(
  task_id = "task01",
  dag = dag,
  sql = "SELECT  * FROM `your_project_id.dataset.table` WHERE dt = '{{ params.target_date }}'",
  gcp_conn_id = "your-conne-id",
  destination_dataset_table = destination_dataset_table)


tasks = [task01]

from_date = datetime.strptime("2021-01-01", '%Y-%m-%d')
to_date = datetime.strptime("2021-04-01", '%Y-%m-%d')



execute_task = ExecuteTaskList(dag, tasks, from_date, to_date, interval)
execute_task.make_task_list()
execute_task.execute_series_task()
```
実装は以上です。
上記コードをデプロイすると下記のようなDAGが作成されます

```sample.py
start_1 >> TEST_PRJ-T001_2021-01-01 >> TEST_PRJ-T001_2021-01-08 >> ... >> end_2
```

# 8. テンプレートパラメーターについて
* 上記サンプルコードではjinjaテンプレートを用いている箇所がありますので簡易フレームワークで使われているjinjaテンプレートを紹介しておきます。
* [Airflow - Templete Parameter](https://airflow.apache.org/docs/apache-airflow/stable/macros-ref.html)

|テンプレート構文|説明|用途|
|:--|:--|:--|
|{{ params.target_date }}|YYYY-MM-DD形式日付に置換|クエリパラメータなどで日付を置き換えたい箇所に使用する|
|{{ params.partition_date }}|YYYYMMDD形式日付に置換|日付パーテションなどで利用|

テンプレートパラメータ利用可能箇所

|利用箇所|可否|備考|
|:--|:--:|:--|
| destination_dataset_table |◯||
| query_params |◯||
| sql |△|パラメーターに直に記述した場合は利用可|