ワークフローエンジン Apache Airflowの理解を深める

本記事は、オープンソースのワークフローエンジンである Apache Airflow の説明と実際のハンズオンを記載する。
AWS では、マネージドサービスとして 2020年11月に Amazon Managed Workflow for Apache Airflow がリリースされたが、そちらは本記事に併せて今後投稿予定。

#Apache Airflowとは
Airflowは、2014年にAirbnb社が開発したオープンソースであり、2016年より Apache財団となる。開発言語は Pythonで、ワークフローエンジンに該当する。

Airflowは、予め決められた順序を基に、処理を実行するワークフローをプログラムで作成する。また、スケジュールや監視を行う事が可能。
ワークフローはタスクの有向非巡回グラフ（DAG）を作成する事により、タスクを実行する。

DAGとは

DAGとは有効非巡回グラフの事で、定義された処理(タスク)をトポロジカルソートによって整理し、ワークフローとして表現するものである。
重要なのは、エッジ(線)のベクトルが有向で、さらに一度通ったノード(点)には戻ってこれないという非巡回性の特性を持つ事である。

##アーキテクチャ
Apache Airflow を理解するために、アーキテクチャを理解していく。

Apache AirFlow Component

AirFlowは主に３つの役割をがコンポーネントとして含まれる。

• Webserver : 画面表示部
• Scheduler : 実行スケジュール管理部
• Executer : 実行部

taskを実行するExecuter、実行スケジュールを管理するScheduler、WebUIでブラウザ上に表示を行うWebserverが存在する。
コンポーネントと大まかな順序は以下のようなイメージ。管理DBを介して 実行スケジュールや依存関係、実行結果を共有する。

Apache AirFlow プログラムアーキテクチャ

Airflow のプログラムは以下のように大きく３つに区分される。

• DAG : どのような順序で実行されるかを記述したプログラム
• Operator : プログラムを実行するためのテンプレート
• Task : 実行する処理

Task の定義に Operator が記述されており、Python で処理を実行するための PythonOperator Bash 実行のための BashOperator、他各種 RDBMS や Hive、AWS や GCP など 様々なサービスの APIをコールする Operator が用意されている。

サンプルコードから処理を見てみる.

dag = DAG(                              //定義
    ‘tutorial’,                         // DAG 名： tutorial
    default_args=default_args,          // パラメーターをデフォルト引数で指定
    schedule_interval=timedelta(days=1) // 実行スケジュール間隔を 1日分として指定
)
t1 = BashOperator(                      // task : t1 を BashOperator で定義
    task_id=‘print_date’,               // task_id を print_date 
    bash_command=‘date’,                // bash  date コマンドの実行
    dag=dag,                            // どの DAG を実行するか定義
)
t2 = BashOperator(                      // task : t2 を BashOperator で定義
task_id='sleep’,                        // task_id を sleep
    depends_on_past=False,              // 指定したタスクの上流タスクの実行が失敗した場合、タスクを実行するかどうかを設定 #False 実行しない
    bash_command=‘sleep 5’,             //　bash sleep 5 の実行
    dag=dag,
)

t2.set_upstream(t1)                     // 実行順序を記載、# t1実行完了後に t2 を実行

###その他機能

その他 Airflow の機能については以下が存在する。

• Connection：各種データストアへの接続情報を管理
• Hook：Connection を使ってデータストアにアクセスしたり、データをload/dumpするためのメソッド
• Pools：タスクの並列数を管理
• Queue：Celeryのような、外部のキューイングシステムをジョブキューとして利用可能
• Branching：DAG中での条件分岐を実現
• SLA：一定時間内に成功しなかったtaskを管理者にメール通知

#導入してみる

導入を行う手順は以下とし、taskのステータスが処理状況が確認出来れば完了とする。

• Dockerコンテナ(Docker for Windows)に Airflowのリポジトリを追加し、WebServerのUIを起動させる。
• ブラウザから Airflowの表示画面を確認出来たら、DAG の作成、Operator、Task を定義する。
• 作成したファイルから再度 Webserverを立ち上げ、DAGが作成されている事を確認し、画面上から Taskの実行を試みる。
• Taskの処理状況を確認する。

##手順

コンテナ上に Airflow のリポジトリ を pullして起動し、ブラウザ上で起動してみたところ、以下のような UIとなる。

> docker pull puckel/docker-airflow
Status: Downloaded newer image for puckel/docker-airflow:latest

> docker run -p 8080:8080 --name airflow puckel/docker-airflow
exec ..

Airflow リポジトリのプロジェクトルートに dags ディレクトリを作成し
$AIRFLOW_HOME/dags 配下に sample.py を配置

述部は割愛するが定義は以下を記載

#DAGの生成 (定義情報) DAG名: first_dag 
#DAGに紐づくタスクの生成  タスク名: t1,t2,t3,t4#タスク間の依存関係を定義
#処理順序  t1の後に t2とt3 を実行、t2・t3 の後にt4が実行
t2.set_upstream(t1)
t3.set_upstream(t1)
t4.set_upstream([t2, t3])

リトライ

Airflow webserver

first_dag という DAG が 生成されている事が確認できる。

また、first_dag を選択すると、Taskが可視化される

実行順序は以下となっている
① t1 実行
② t2 および t3 の実行
③ t2 および t3 の実行が完了した後に t4 の実行

実行形式は以下となる
Toggleを onにクリック (定刻実行 )
Trigger Dagをクリック (即時実行)

実行状況は taskフローから確認する事が出来た。