Apache Airflow触ってみた - 名もなき未知

アドベントカレンダー14 日目です（大幅遅刻）https://adventar.org/calendars/8021

さて、apache/airflow： Apache Airflow - A platform to programmatically author, schedule, and monitor workflows とかあったねーと思ったので、軽く動かしてみました。

環境の準備

What is Airflow? — Airflow Documentation

こちらのページを軽めに読んでいる感じ、パスや設定に強く依存している印象を受けたので、Docker 上で構築することにしました。

公式に用意されている docker-compose.yml を使ってもよかったのですが、初手で DB とか Redis とか使わなくてもいいかなと思ったので、簡単な Dockerfile にしました。

FROM python:3.10

WORKDIR /root
RUN export AIRFLOW_HOME=/root
RUN mkdir -p /root/airflow/dags
COPY requirements.txt .

RUN pip install -r requirements.txt
COPY bash_dag.py ./airflow/dags/

CMD ["airflow", "standalone"]

動作させるスクリプト

AIRFLOW_HOME を意識したうえで適切なパスに置きます。今回は /root を HOME にしているので、 /root/airflow/dags に置きました。

Bash の echo とか動けばいいなーくらいのものを目標にして進めてみました。コードサンプルとかいろいろありましたが、 ChatGPT で適当なコードを書いてもらったので、それを少し編集する形で下記のモノを作りました。

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta

# このDAGのIDを指定します
dag_id = 'bash_dag'

# このDAGを開始する日時を指定します
start_date = datetime(2022, 1, 1)

# DAGを定義します
dag = DAG(
    dag_id,
    start_date=start_date,
    catchup=False
)

# 最初のタスクを定義します
task1 = BashOperator(
    task_id='task1',
    bash_command='echo "Task 1"',
    dag=dag
)

# 2番目のタスクを定義します
task2 = BashOperator(
    task_id='task2',
    bash_command='echo "Task 2"',
    dag=dag
)

# 3番目のタスクを定義します
task3 = BashOperator(
    task_id='task3',
    bash_command='echo "Task 3"',
    dag=dag
)

# 最初のタスクが完了した後に、2番目, 3番目のタスクを実行するように依存関係を設定します
task1 >> [task2, task3]

実際にWeb UIを使ってみる

GUI から実行してみます。（ちなみに admin の初期パスは設定もできるらしいのですが、今は standalone_admin_password.txt を見ています）

適当に build したら、手軌道に下記のコマンドで起動します（常に起動しててよいので -d オプションをしています）

docker run -d -p 8080:8080 airflow-tutorial:lates

admin パスワードは下記のコマンドで確認します。

docker exec -it $CONTAINER_ID cat airflow/standalone_admin_password.txt

あとはいろいろ実行してみて（この画面に行って▷で実行）

実行結果を眺めたりするのがいいのかも。

ログもわかりやすくていいですね。

まとめ

ざっくり動かしてみたのですが、結構いい感じに可視化してくれたり、コードで実行するものを選べたりといい感じですね。

途中から実行を再開する機能や、他にも使えていないもの（今回は BashOperator を使ったが、他にもいろいろな Operator があるらしい）もあるので、思いつくものはいくつか試してみたいなーと思います。（自分の考えている用途だとバックアップ系とかになっちゃうかなーと思います、個人でデータパイプラインを構築してあれこれというのはあんまりやらないかも）

面白そうなのでまた深堀してみたいなと思います。