Airflowは、Pythonベースのジョブスケジューリングツールであり、大規模なデータ処理やETLプロセスのスケジュールに最適です。Airflowは、依存関係のあるジョブのワークフローを定義し、スケジュールするための高度な機能を提供します。また、Airflowは、WebUIを介してジョブのステータスを監視し、ジョブの実行履歴を追跡することができます。この記事では、Airflowを使ってジョブスケジューリングを行う方法を説明します。以下の手順に従って、Airflowを使って簡単にジョブスケジュールを作成することができます。
1. Airflowのインストール
Airflowは、pipを使ってインストールすることができます。以下のコマンドを使用して、Airflowをインストールしてください。
pip install apache-airflow
2. Airflowの初期化
Airflowを初期化するには、以下のコマンドを使用して、Airflowのメタデータを初期化する必要があります。
airflow initdb
3. DAG（Directed Acyclic Graph）の作成
Airflowでジョブスケジューリングを行うには、まず、DAGを定義する必要があります。DAGは、依存関係のあるタスクのグラフを表します。DAGはPythonファイルとして記述し、AirflowのDAGフォルダに保存する必要があります。以下は、サンプルのDAGファイルの例です。
Bashコマンドを実行するDAGの例
下記は2つのBashコマンドを実行するDAGファイルの例です。
from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash_operator import BashOperator default_args = { 'owner': 'airflow', 'depends_on_past': False, 'start_date': datetime(2022, 1, 1), 'email': ['airflow@example.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'my_dag', default_args=default_args, description='A simple tutorial DAG', schedule_interval=timedelta(days=1), ) t1 = BashOperator( task_id='task_1', bash_command='echo "Hello World"', dag=dag, ) t2 = BashOperator( task_id='task_2', bash_command='echo "Hello Airflow"', dag=dag, ) t1 >> t2
このDAGファイルでは、2つのBashOperatorを定義しています。それぞれのタスクは、Bashコマンドを実行して、"Hello World"と"Hello Airflow"というメッセージを出力します。また、2つのタスクの間には、依存関係を設定するために、"t1 >> t2"という行があります。これにより、t2タスクはt1タスクが完了する前に開始されません。
Pythonスクリプトを実行するDAGの例
PythonのスクリプトをAirflowを使って定期的に実行するDAGを定義してみます。下記の様にDAGを定義します。このDAGは、1日に1回実行され、script1.pyとscript2.pyが交互に実行されます。
from datetime import datetime, timedelta from airflow import DAG from airflow.operators.bash_operator import BashOperator default_args = { "owner": "airflow", "depends_on_past": False, "start_date": datetime(2023, 4, 22), "retries": 0, "retry_delay": timedelta(minutes=5), } dag = DAG( "sample_dag", default_args=default_args, description="Sample DAG to run Python scripts", schedule_interval=timedelta(days=1), ) script1 = BashOperator( task_id="script1", bash_command="python /path/to/project/scripts/script1.py --name Airflow", dag=dag, ) script2 = BashOperator( task_id="script2", bash_command="python /path/to/project/scripts/script2.py --num 10", dag=dag, ) script1 >> script2
このDAGを実行すると、script1.pyが実行され、Hello, Airflow!というメッセージが表示されます。その後、script2.pyが実行され、The sum of numbers from 1 to 10 is 55.というメッセージが表示されます。このように、Airflowを使ってPythonのスクリプトを定期的に実行することができます。DAGの定義によって、より複雑なスケジュールやジョブ間の依存関係を定義することもできます。ただし、DAGに含まれるタスクが実行される際には、各タスクが依存するライブラリやパッケージが正しくインストールされていることを確認する必要があります。また、環境変数やファイルパスなど、各タスクが必要とする情報を正しく設定することも重要です。
4. DAGの実行
DAGを実行するには、以下のコマンドを使用します。
airflow scheduler
このコマンドを実行すると、Airflowのスケジューラが起動し、DAGが定義されたスケジュールに従って実行されます。また、WebUIからDAGのステータスを監視することもできます。以下のURLにアクセスすると、AirflowのWebUIにアクセスできます。
http://localhost:8080
まとめ
AirflowのWebUIでは、DAGの実行履歴やタスクのステータスを確認することができます。また、タスクの実行ログを表示することもできます。まとめAirflowを使ってジョブスケジューリングを行う方法について説明しました。Airflowは、依存関係のあるタスクのスケジュールを簡単に管理することができるため、大規模なデータ処理やETLプロセスのスケジュールに最適です。DAGを定義することで、複雑なジョブのスケジュールを簡単に管理することができます。また、AirflowのWebUIから、ジョブのステータスやログを簡単に監視することができます。Pythonでの業務の自動化について学ぶには下記のようなサイトが有効です。
