ワークフローエンジン Apache Airflowの理解を深める

**本記事は、オープンソースのワークフローエンジンである Apache Airflow の説明と実際のハンズオンを記載する。**
**AWS では、マネージドサービスとして 2020年11月に Amazon Managed Workflow for Apache Airflow がリリースされたが、そちらは本記事に併せて今後投稿予定。**

#Apache Airflowとは
Airflowは、2014年にAirbnb社が開発したオープンソースであり、2016年より Apache財団となる。開発言語は Pythonで、ワークフローエンジンに該当する。

https://airflow.apache.org/

Airflowは、予め決められた順序を基に、処理を実行するワークフローをプログラムで作成する。また、スケジュールや監視を行う事が可能。
ワークフローはタスクの有向非巡回グラフ（DAG）を作成する事により、タスクを実行する。

## DAGとは
DAGとは有効非巡回グラフの事で、定義された処理(タスク)をトポロジカルソートによって整理し、ワークフローとして表現するものである。
**重要なのは、エッジ(線)のベクトルが有向で、さらに一度通ったノード(点)には戻ってこれないという非巡回性の特性を持つ事である。**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/ab092dce-14ef-42b6-5cdc-bffd71394681.png)

##アーキテクチャ
Apache Airflow を理解するために、アーキテクチャを理解していく。

### Apache AirFlow Component
AirFlowは主に３つの役割をがコンポーネントとして含まれる。

* <font color="Red">Webserver</font> : 画面表示部 
* <font color="Blue">Scheduler</font> : 実行スケジュール管理部 
* <font color="Green">Executer</font> : 実行部 

taskを実行する<font color="Green">Executer</font>、実行スケジュールを管理する<font color="Blue">Scheduler</font>、WebUIでブラウザ上に表示を行う<font color="Red">Webserver</font>が存在する。
コンポーネントと大まかな順序は以下のようなイメージ。管理DBを介して 実行スケジュールや依存関係、実行結果を共有する。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/48b2bc77-ef88-7daf-46ba-469ff4b6edbd.png)

### Apache AirFlow プログラムアーキテクチャ
Airflow のプログラムは以下のように大きく３つに区分される。

* <font color="Red">DAG</font> :  どのような順序で実行されるかを記述したプログラム
* <font color="Blue">Operator</font> :  プログラムを実行するためのテンプレート
* <font color="Green">Task</font> : 実行する処理

<font color="Green">Task</font> の定義に <font color="Blue">Operator</font> が記述されており、Python で処理を実行するための PythonOperator Bash 実行のための BashOperator、他各種 RDBMS や Hive、AWS や GCP など 様々なサービスの APIをコールする Operator が用意されている。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/befc1c6d-c533-5915-abe4-725c5d5db272.png)

サンプルコードから処理を見てみる.

```

dag = DAG(                              //定義
    ‘tutorial’,                         // DAG 名： tutorial
    default_args=default_args,          // パラメーターをデフォルト引数で指定
    schedule_interval=timedelta(days=1) // 実行スケジュール間隔を 1日分として指定
)
t1 = BashOperator(                      // task : t1 を BashOperator で定義
    task_id=‘print_date’,               // task_id を print_date 
    bash_command=‘date’,                // bash  date コマンドの実行
    dag=dag,                            // どの DAG を実行するか定義
)
t2 = BashOperator(                      // task : t2 を BashOperator で定義
task_id='sleep’,                        // task_id を sleep
    depends_on_past=False,              // 指定したタスクの上流タスクの実行が失敗した場合、タスクを実行するかどうかを設定 #False 実行しない
    bash_command=‘sleep 5’,             //　bash sleep 5 の実行
    dag=dag,
)

t2.set_upstream(t1)                     // 実行順序を記載、# t1実行完了後に t2 を実行

```

###その他機能

その他 Airflow の機能については以下が存在する。

* Connection：各種データストアへの接続情報を管理
* Hook：Connection を使ってデータストアにアクセスしたり、データをload/dumpするためのメソッド
* Pools：タスクの並列数を管理
* Queue：Celeryのような、外部のキューイングシステムをジョブキューとして利用可能
* Branching：DAG中での条件分岐を実現
* SLA：一定時間内に成功しなかったtaskを管理者にメール通知

#導入してみる

導入を行う手順は以下とし、taskのステータスが処理状況が確認出来れば完了とする。

* Dockerコンテナ(Docker for Windows)に Airflowのリポジトリを追加し、WebServerのUIを起動させる。
* ブラウザから Airflowの表示画面を確認出来たら、DAG の作成、Operator、Task を定義する。
* 作成したファイルから再度 Webserverを立ち上げ、DAGが作成されている事を確認し、画面上から Taskの実行を試みる。
* Taskの処理状況を確認する。

##手順

コンテナ上に Airflow のリポジトリ を pullして起動し、ブラウザ上で起動してみたところ、以下のような UIとなる。

```
> docker pull puckel/docker-airflow
Status: Downloaded newer image for puckel/docker-airflow:latest

> docker run -p 8080:8080 --name airflow puckel/docker-airflow
exec ..
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/1338d9f4-f45b-dae4-886c-6aa35b502baf.png)

Airflow リポジトリのプロジェクトルートに dags ディレクトリを作成し
$AIRFLOW_HOME/dags 配下に sample.py を配置

述部は割愛するが定義は以下を記載 

```
#DAGの生成 (定義情報) DAG名: first_dag 
#DAGに紐づくタスクの生成  タスク名: t1,t2,t3,t4#タスク間の依存関係を定義
#処理順序  t1の後に t2とt3 を実行、t2・t3 の後にt4が実行
t2.set_upstream(t1)
t3.set_upstream(t1)
t4.set_upstream([t2, t3])
```
リトライ

```
Airflow webserver
```
first_dag という DAG が 生成されている事が確認できる。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/0684151f-a14f-0228-04f3-4ac0b197a66b.png)

また、first_dag を選択すると、Taskが可視化される

実行順序は以下となっている
① t1 実行
② t2 および t3 の実行
③ t2 および t3 の実行が完了した後に t4 の実行
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/cdf0942f-7c9b-a8a4-958f-47a1f62cd843.png)

実行形式は以下となる
Toggleを onにクリック (定刻実行 ) 
Trigger Dagをクリック (即時実行) 

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/26e4552c-8fcb-2c21-f852-445834186636.png)

実行状況は taskフローから確認する事が出来た。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1597898/eb423cfb-975e-a7bd-cf03-cee9e67d8d00.png)