Airflow のアーキテクチャをざっくり理解して、どうやって使うのか学んでみた

ジョブ管理ツール

みかみ

2020.03.05

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、みかみです。

Python で実装されている Job 管理ツール Apache Airflow。 WebUI のJob 管理画面で直感的に分かりやすい操作が可能で、エラー発生時などの通知機能もあり、スケールアウトにも対応していて複数サーバで分散実行もできます。

Python でバッチ Job 開発経験のある方には多分おなじみの Airflow、私も存在は知っていましたが、実際使ったことはありませんでした。

やりたいこと

Airflow の構成（アーキテクチャ）を知りたい
Airflow の使い方（Job 作成＆実行方法）を知りたい

Airflow のアーキテクチャ

Airflowは、管理画面表示部の Webserver と、Job実行のスケジュール管理部の Scheduler 、Job実行部の Worker(Executer) から成り立っているようです。

各モジュールは管理 DB を介して Job の実行スケジュールや依存関係、実行結果を共有します。

Webserver は Python の Web アプリケーションフレームワーク Flask で実装されています。

また Worker(Executer) は、例えば Python の分散タスクキューフレームワーク Celery を使用した分散並列処理などにカスタマイズすることもできます。

Job Queue を使わない構成や管理 DB のエンジンも自由に選択可能なのであくまでイメージですが、ざっくり以下のような構成になっていると理解しました。

Airflow の Job 構成

Airflow では、Job の実行順や依存関係などのワークフローは DAG で定義するとのこと。

Job本体（実際の処理）は Operator に記述されており、Python で処理を実行するための PythonOperator や、Bash 実行のための BashOperator の他、各種 RDBMS や Hive、AWS や GCP など様々なサービスに対する処理を実行する Operator も用意されています。

さらに、 BaseOperator クラスを継承して、必要な処理を実行する Operator を自由に追加することもできます（カスタマイズ性高くて良いですねｖ

DAG を追加して動かしてみる

と、学んだところで、実際に Airflow で Job を作成して動かしてみようと思います。

まずは Airflow をインストール。 Python 3.7 実行環境構築済みの EC2（Amazon Linux）に、下記クイックスタートに記載のコマンドでインストール＆起動しました。

Quick Start | Apache Airflow Documentation

管理画面にアクセスしてみると、サンプル DAG のリスト表示が確認できました。

続いて、チュートリアルとサンプルの example_python_operator.py を参考に、DAG を追加してみます。

$AIRFLOW_HOME 配下に dags ディレクトリを作成して、test_mikami.py ファイルを追加しました。

(test_airflow) [ec2-user@ip-10-0-43-239 airflow]$ pwd
/home/ec2-user/airflow
(test_airflow) [ec2-user@ip-10-0-43-239 airflow]$ mkdir dags
(test_airflow) [ec2-user@ip-10-0-43-239 airflow]$ cd dags
(test_airflow) [ec2-user@ip-10-0-43-239 dags]$ vi test_mikami.py

BashOperator で date コマンドを実行してから5秒間 sleep し、PythonOperator で「Hello! Airflow!!」と出力するだけの単純な処理です。

from datetime import timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago

from pprint import pprint
from airflow.operators.python_operator import PythonOperator

# Parameters
default_args = {
    'owner': 'ec2-user',
    'depends_on_past': False,
    'start_date': days_ago(2),
}
dag = DAG(
    'test_mikami',
    default_args=default_args,
    description='For TEST execute DAG',
    schedule_interval=timedelta(days=1),
)

# BashOperator：date + sleep 5
t1 = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag,
)

t2 = BashOperator(
    task_id='sleep',
    depends_on_past=False,
    bash_command='sleep 5',
    retries=3,
    dag=dag,
)

# PythonOperator：print
def print_context(ds, **kwargs):
    pprint(kwargs)
    print(ds)
    return 'Hello! Airflow!!'


run_this = PythonOperator(
    task_id='print_the_context',
    provide_context=True,
    python_callable=print_context,
    dag=dag,
)

t1 >> t2 >> run_this

airflow webserver を再起動してみると、管理画面のDAG リストにも、test_mikami が追加表示されました。