Apache Airflow 並列処理 ハンズオン

# はじめに

[Apache Airflow ハンズオン](https://qiita.com/honda28/items/2d780aad39b344e354ff)に続いて、[The Complete Hands-On Introduction to Apache Airflow](https://www.udemy.com/course/the-complete-hands-on-course-to-master-apache-airflow/) で理解を深める。本記事では、デフォルトで設定されている SQLite と SequentialExecutor での挙動を確認した後、それらを PostgreSQL と LocalExecutor に変更し、並列処理を試す。

# デフォルト設定の確認

Apache Airflow では、初期化した段階でデフォルトの設定がなされている。設定は以下コマンドで確認できる。

```bash
$ airflow config get-value core sql_alchemy_conn
sqlite:////home/airflow/airflow/airflow.db
$ airflow config get-value core executor
SequentialExecutor
```

これを見ると、DB には SQLite が採用され、Executor は SequentialExecutor となっていることがわかる。一旦、デフォルトの設定のままでどのような挙動になるかを確認する。

以下で簡単な DAG を作成した。

```python:dags/parallel_dag.py
from datetime import datetime

from airflow import DAG
from airflow.operators.bash import BashOperator

default_args = {"start_date": datetime(2020, 1, 1)}

with DAG(
    "parallel_dag", schedule_interval="@daily", default_args=default_args, catchup=False
) as dag:

    task_1 = BashOperator(task_id="task_1", bash_command="sleep 3")

    task_2 = BashOperator(task_id="task_2", bash_command="sleep 3")

    task_3 = BashOperator(task_id="task_3", bash_command="sleep 3")

    task_4 = BashOperator(task_id="task_4", bash_command="sleep 3")

    task_1 >> [task_2, task_3] >> task_4
```

`airflow webserver` と `airflow scheduler` を実行し、[http://localhost:8080](http://localhost:8080) にアクセス。DAGs の中から `parallel_dag` を選択し、Graph View を選ぶと以下のような画面となる。

![parallel_dag.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/769545/205bef04-e22b-e480-f91e-cad6e93982a1.png)

このグラフを見ると、並列処理が実装されていそうだが、画面左上のトグルをオンにして実行した後に Gantt 画面に遷移すると並列処理ではなく、順次実行されていることが確認できる。

![gantt.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/769545/1bddac38-11ae-fe5b-e277-ab72680bfccf.png)

これはデフォルトの Executor が Sequantial Executor であることが原因である。これら設定は `~/airflow/airflow.cfg` に記述されている。

# 並列処理を行うための設定

並列処理を行うためには、並列処理に対応する DB と Executor に変更する必要がある。ここでは、DB として PostgreSQL を採用し、Executor を Sequantial Executor から Local Executor に変更する。

```bash
$ sudo apt update
$ sudo apt install postgresql
```

接続するために、ユーザ postgres のパスワードを設定。

```bash
$ sudo -u postgres psql
postgres=# ALTER USER postgres PASSWORD 'postgres';
ALTER ROLE
```

Postgres を扱うためのパッケージをインストール。

```bash
$ pip install 'apache-airflow[postgres]'
```

以下で設定を変更していく。まず DB 設定を以下のように変更した。設定の正確なフォーマットは [Database Urls](https://docs.sqlalchemy.org/en/14/core/engines.html#database-urls) を参照。

```~/airflow/airflow.cfg
...

# sql_alchemy_conn = sqlite:////home/airflow/airflow/airflow.dbost:port/database
sql_alchemy_conn = postgresql+psycopg2://postgres:postgres@localhost/postgres
```

変更を反映した後、以下コマンドが問題なく実行されれば、DB を PostgreSQL に設定できている。

```bash
$ airflow db check
```


続いて、Executor も変更する。

```~/airflow/airflow.cfg
...

# executor = SequentialExecutor
executor = LocalExecutor
```

設定を変更できたので、Postgres DB を初期化する。`airflow webserver` と `airflow scheduler` が立ち上がっている場合は、一旦停止しておく。

```bash
$ airflow db init
...
$ airflow users create -u admin -p admin -r Admin -f admin -l admin -e admin@airflow.com
...
Admin user admin created
```


DB の初期化および新しいユーザの作成を行ったので、`airflow webserver` と `airflow scheduler` を実行し、[http://localhost:8080](http://localhost:8080) にアクセス。以前のユーザでログインしたままの場合は、一度ログアウトして admin/admin でログインし直す。DAGs の中から `parallel_dag` を選択し、Graph View を選ぶと以下のような画面となる。

![parallel_gantt.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/769545/d15030d6-d828-d8b5-d42a-267b67c1c04b.png)

並列処理されていることが確認できる。

# おわりに

PostgreSQL と LocalExecutor を用いて、並列処理を実行してみた。引き続き、理解を深めていきたい。