Apache Airflowのチュートリアルをローカル環境で試す

# Apache Airflow　2.2.3
[<img width = 250 src = "https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2296212/fe59cc46-a5cf-ade4-8f72-a17bd9a17ee3.png">](https://airflow.apache.org/docs/apache-airflow/stable/index.html)

## Airflowとは
Apache AirflowはAirbnbのMaxime Beauchemin氏によって開始されたPython実装によるJob管理ツールで、ワークフローをプログラムで作成/スケジューリング/監視するためのプラットフォーム。
AirflowではJobの実行順だったり、依存関係のワークフローはDAG（Directed Acyclic Graph）へ定義する。

## アーキテクチャ
[<img src = "https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2296212/9ad8c87d-fbae-ac67-1f41-06f4c53d85f6.png">](https://airflow.apache.org/docs/apache-airflow/stable/concepts/overview.html)

Airflowは下記のコンポーネントで構成される。
* スケジュールされたワークフローのトリガーと、実行タスクをExecuterの送信するための**Scheduler**。
* 実行中のタスクを処理する**Executer**。デフォルトではスケジューラ内ですべてを実行するが、本番環境に適したExecuterはタスクの実行をWorkersにプッシュする。
* DAGおよびタスクの動作を検査、トリガー、およびデバッグするための便利なユーザーインターフェイスを提供する**Webサーバー**。
* Scheduler/ Executorに読み取られる**DAGファイルのフォルダ**。
* 状態保持のために　Scheduler/ Executor/ Webサーバー　に使用される**Meta DB**。

## 実行準備

Airflowのパスを設定。

```console
$ export AIRFLOW_HOME=~/airflow
```

Airflowインストールコマンド。

```
$ pip install "apache-airflow[celery]==2.2.3" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.2.3/constraints-3.6.txt"
```

ここでは推奨されているPostgreSQLでDBを準備

```sql
CREATE DATABASE airflow_db;
CREATE USER airflow_user WITH PASSWORD 'airflow_pass';
GRANT ALL PRIVILEGES ON DATABASE airflow_db TO airflow_user;
```

Airflowでは初期DBがSQLiteになっているらしいので、
airflow/airflow.cfgを編集して

`sql_alchemy_conn = sqlite:////home/user/airflow/airflow.db`を
`sql_alchemy_conn = postgresql://airflow_user:airflow_pass@localhost/airflow_db`へ変更

DBの初期化やユーザ作成を実行。

```console
$ airflow standalone
```

### コードの準備

airflow/dagsのフォルダを作成してその下にtest-tutorial.pyを作成して中身は
チュートリアルページのtutorial.pyをパクってDAGを作成。

```python
from datetime import datetime, timedelta
from textwrap import dedent

# The DAG object; we'll need this to instantiate a DAG
from airflow import DAG

# Operators; we need this to operate!
from airflow.operators.bash import BashOperator
# These args will get passed on to each operator
# You can override them on a per-task basis during operator initialization
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
    # 'wait_for_downstream': False,
    # 'dag': dag,
    # 'sla': timedelta(hours=2),
    # 'execution_timeout': timedelta(seconds=300),
    # 'on_failure_callback': some_function,
    # 'on_success_callback': some_other_function,
    # 'on_retry_callback': another_function,
    # 'sla_miss_callback': yet_another_function,
    # 'trigger_rule': 'all_success'
}
with DAG(
    'test-tutorial',
    default_args=default_args,
    description='A simple test-tutorial DAG',
    schedule_interval=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['example'],
) as dag:

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    t1 = BashOperator(
        task_id='print_date',
        bash_command='date',
    )

    t2 = BashOperator(
        task_id='sleep',
        depends_on_past=False,
        bash_command='sleep 5',
        retries=3,
    )
    t1.doc_md = dedent(
        """\
    #### Task Documentation
    You can document your task using the attributes `doc_md` (markdown),
    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
    rendered in the UI's Task Instance Details page.
    ![img](http://montcs.bloomu.edu/~bobmon/Semesters/2012-01/491/import%20soul.png)

    """
    )

    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # otherwise, type it like this
    templated_command = dedent(
        """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 7)}}"
        echo "{{ params.my_param }}"
    {% endfor %}
    """
    )

    t3 = BashOperator(
        task_id='templated',
        depends_on_past=False,
        bash_command=templated_command,
        params={'my_param': 'Parameter I passed in'},
    )

    t1 >> [t2, t3]
```

#### コードを紐解く

|パラメータ|型|説明|
|---|---|---|
|dag_id	|str|DAG単位で与える一意な識別子。ここでは'test-tutorial'が該当。|
|default_args| dict |オペレータに渡す引数。引数の中には優先順位があり、優先順から<br>1. 明示した引数<br>2. default_args<br>3. オペレータのデフォルト値（存在する場合)<br>となる。|


### executorについて
作成したDAGがなぜか読み込まれなかったが、
airflow/airflow.cfg内のexecutorをSequentialExecutorからCeleryExecutor変更することで読み込まれた（SequentialExecutorではタスクを直列にしか実行できないから？）。

executerについて下記にまとめる
| SequentialExecuter | CeleryExecutor | LocalExecutor |
| -- | -- | -- |
| デフォルトのExecutorで、直列にしかタスクを実行できない。お試し版的存在 | キューを利用したExecutorでスケジューラとworkerが分離される。 | schedulerとworkerを同じノードで構成する場合に使用するExecutor |
| スケールは一切しない | workerを並列実行でき、スケールアップおよびスケールアウト でスケール可能になるため本番環境向き。 | マルチプロセスで動作し、スケールアップ によってスケール可能 |


## 実行

利用するコマンド

```console
$ airflow webserver
```

実行すると下記のような画面がlocalhost:8080で開く

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2296212/7aa32fae-39e2-ba71-2cea-db2d7ce16a07.png)


admin/adminで入れなければログインするためのユーザを作成してログイン

```
$ airflow users  create --role Admin --username admin --email admin --firstname admin --lastname admin --password admin
```

DAGsからtest-tutorialを選択。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2296212/4a98de2a-f669-9d34-214f-95e3d78d535d.png)

Trigger DAGで手動実行
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2296212/a5f943fd-13e1-3d93-415f-021875debb8c.png)

実行ステータスが視覚的に確認できる
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2296212/85b8ff01-a302-b1f6-5283-875b706e168c.png)

ログも出ていることを確認
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2296212/1ef708ba-a5b2-c30f-17bc-4a87d72ad106.png)

## 用語
### タスク
タスクは、オペレータオブジェクトをインスタンス化する際に生成される。演算子からインスタンス化されたオブジェクトをタスクと呼んでいる。ソースで言うと以下。

```python
t1 = BashOperator(
    task_id='print_date',
    bash_command='date',
)
```

第1引数のtask_idは、タスクの一意な識別子として機能する。



## 参考
* [2018 builderscon airflowを用いて、 複雑大規模なジョブフロー管理 に立ち向かう](https://www.slideshare.net/BrainPad/2018-builderscon-airflow)
* [Apache Airflowの並列性と並行性について理解する](https://dev.classmethod.jp/articles/apache-airflow-parallelism-and-concurrency/)
* [Airflow を用いたデータフロー分散処理](https://analytics.livesense.co.jp/entry/2018/02/06/132842)