airflowのチュートリアルをしてみた

##背景
業務でairflowを触るかもしれないので、事前にいろいろ調べておこうと思ったのがキッカケです。
ただドキュメントを読むだけだとあまりイメージわかないと思うので実際に手を動かしながらやってみます。
参考にしたのはこちらのサイトです。この場を借りてお礼申し上げます。
https://ohke.hateblo.jp/entry/2018/04/21/230000

##準備
普通にローカルにairflowをインストールできますが、せっかくなのでdockerでやってみます。
こちらのリポジトリを使います。
https://github.com/puckel/docker-airflow
さっそくcloneしていきます。
コマンドはこれで。

```
git clone git@github.com:puckel/docker-airflow.git
```

ついでにairflowの追加パッケージやpythonの依存関係を解決するために、
cloneしたリポジトリ配下に行き、以下のコマンドを実行しておきます。

```
docker build --rm --build-arg AIRFLOW_DEPS="datadog,dask" --build-arg PYTHON_DEPS="flask_oauthlib>=0.9" -t puckel/docker-airflow .
```

ちょっと時間がかかるので、この間にコーヒーとかお茶を用意して待ちましょう。
ソシャゲのAPを消費しても良いかもしれません。

終わったら以下のコマンドを実行してairflowを立ち上げます。

```
docker run -d -p 8080:8080 puckel/docker-airflow webserver
```

うまく行けばこのような画面が表示されるかと思います。こうなればひとまず準備完了です。
<img width="1440" alt="スクリーンショット 2019-05-19 13.47.37.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/100993/f506b702-5532-4f0b-9eb3-df9764071c40.png">
ここまで大体10分かかっていないです。楽に用意できて良いですね！

##DAGの準備
環境も整ったので、実際にDAGを設定していきましょう。
DAGとは「有効非巡回グラフ(Directed acyclic graph)」の略で、
airflowでは複数集まったタスクのまとまりのことを言います。（詳しくは[wikipedia](https://ja.wikipedia.org/wiki/%E6%9C%89%E5%90%91%E9%9D%9E%E5%B7%A1%E5%9B%9E%E3%82%B0%E3%83%A9%E3%83%95)からどうぞ）

とりあえずDAGに関しては元々入っていた`tuto.py`を使ってみます。
DAGの定義はpythonで行い、作成したDAGは/dagsに配置するとDAGとして認識し実行できるようになります。
（この設定は`airflow.cfg`で変えられます。）

###しかし...
ここで躓きました。。。
docker-airflowのREADMEと背景で紹介したサイトを頼りに動かそうとしていたのですが、airflowを実行する時に```SequentialExecutor```で動かしていました。
これでDAGが表示されるかと思ったのですが、全く表示されず。。。
いろいろ調べていたら以下の記事が。。。
https://qiita.com/kysnm/items/feda7b8cca44bb7389ac
どうも```SequentialExecutor```ではうまく動かないっぽく手を加える必要があります。
さらにこれとは別で紹介されている```LocalExecutor```にはジョブを実行するためのschedulerが入っていないという・・・
直すのも手間なので、すでに動いている実績がある```CeleryExecutor```を使うことになりました。

そのため、元々立ち上がっていたコンテナをkillして以下のコマンドでもう一度立ち上げます。
```docker-compose -f docker-compose-CeleryExecutor.yml up -d```
すると以下のような画面になるかと思われます。
<img width="1440" alt="スクリーンショット 2019-05-19 14.47.58.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/100993/53f814ca-3c05-77a9-6e0e-392ae42a74f6.png">
ここまでくれば一安心です。一旦休憩しましょう。

##実際にDAGを作ってみる
ちょっと休憩したので実際にDAGを作ってみます。
といってもそんな高度なものではなく簡単なものです。
作成する場所は`/dags`以下に作ります。

```python:Hello.py
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta


default_args = {
    "owner": "raqwel",
    "depends_on_past": False, # 上流タスクが失敗した時に、タスクを実行するかどうか
    "start_date": datetime(2019, 5, 19),
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    dag_id="dag_test",
    default_args=default_args,
    catchup=False, # タスク実行に期間が空いてしまった場合、その期間の分のタスクを実行するかどうか
    schedule_interval=timedelta(1)
)

t1 = BashOperator(task_id="print_hello", bash_command="echo Hello World", dag=dag)

t2 = BashOperator(task_id="wait", bash_command="sleep 5", dag=dag)


template_command = """
    echo "hello airflow"
"""

t3 = BashOperator(task_id="template_hello", bash_command=template_command, dag=dag)

t2.set_upstream(t1)
t3.set_upstream(t2)
```
はい、tuto.pyを写経して少し変えてみただけのファイルです。

###確認してみる
画面に戻ると先程作成したDAGが登録されているのがわかります。
<img width="1438" alt="スクリーンショット 2019-05-19 15.37.42.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/100993/133a22d0-20bd-f37b-6ddc-be230c843ed6.png">
何もしてないけど読み込んだ・・・すごい・・・
ということで左のOffになっているスイッチをOnにします。
Graph_viewを見てみると自分が設定した順にtaskを実行するようになっています。
<img width="1434" alt="スクリーンショット 2019-05-19 15.39.43.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/100993/d7bf2c90-ebe1-b1bc-c207-a56565b8458e.png">
とりあえず設定は完了した模様です。これで時間が来たら設定したとおりに実行してくれます。
ですが、今回は実行結果を早く見たいので、自分で実行してみます。
Linksの左にある再生ボタンをクリックしたら「本当に実行しますか」と画面に出てくるのでOKを押します。
<img width="319" alt="スクリーンショット 2019-05-19 15.44.46.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/100993/b9eec08f-bb7a-0523-4f4e-de7498bde383.png">
これで実行されます。そしてログを見に行きます。
```Graph View```タグから適当にtaskを一つクリックし、
出てくるポップアップ内にあるView Logをクリックします。
今回は```print_hello```を選択します。
すると以下のようなログがある画面が出てきます
<img width="1429" alt="スクリーンショット 2019-05-19 15.48.57.png" src="https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/100993/304b8e09-6c41-e7a9-b716-37ce84f5e498.png">
見てみると```Running command```のあとに実行された結果が載っています。うまく実行できているみたいです。

###まとめ
今回はairflowを少し触ってみました。
使ってみて良いなと思ったのは、やはりGUI部分でしょうか。
設定したタスクが成功しているか失敗しているか、どのタスクがどのタスクに依存していて、どれだけの時間実行されていたかがぱっと見てわかるのは強みだと感じました。
またいろいろ上流タスクの成否、空き期間時による実行制御等細かく設定できるのも役に立ちそうです。
今回はBashのオペレーターを使いましたが、他にもPythonなど色々な言語で処理を定義できるので、色々試していきたいと思います。
本当はここからECSも使いたかったのですが、疲れたのでここまでにします。