AirflowのDAGとは？定義方法と最適な設計パターンについて解説！

# DAGの定義（ワークフロー全体の管理）
with DAG(
    'example_dag',  # DAG名（Airflow UI上での識別子）
    start_date=datetime(2023, 1, 1),  # DAGの開始日時
    schedule_interval='@daily',  # スケジュール（毎日実行）
    catchup=False  # 過去分の実行をスキップ
) as dag:
    
    # データを取得する関数
    def extract_data():
        print("データを取得しています...")
    
    # データを変換する関数
    def transform_data():
        print("データを変換しています...")
    
    # データを保存する関数
    def load_data():
        print("データを保存しています...")
    
    # 各処理をPythonOperatorとして定義（Airflow内で実行されるタスク）
    extract_task = PythonOperator(
        task_id='extract_data',  # タスクID（Airflow上での識別名）
        python_callable=extract_data  # 実行するPython関数
    )
    
    transform_task = PythonOperator(
        task_id='transform_data',
        python_callable=transform_data
    )
    
    load_task = PythonOperator(
        task_id='load_data',
        python_callable=load_data
    )
    
    # タスクの依存関係を設定（実行順序を定義）
    extract_task >> transform_task >> load_task

def process_data():
    print("Processing data")

def validate_data():
    print("Validating data")

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
from tasks import process_data, validate_data

with DAG('modular_dag', start_date=datetime(2023, 1, 1), schedule_interval='@daily', catchup=False) as dag:
    task1 = PythonOperator(task_id='process_data', python_callable=process_data)
    task2 = PythonOperator(task_id='validate_data', python_callable=validate_data)

    task1 >> task2  # タスクの依存関係

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime

# DAGを動的に生成する関数
def generate_dag(dag_id):
    with DAG(dag_id, start_date=datetime(2023, 1, 1), schedule_interval='@daily', catchup=False) as dag:
        task = BashOperator(
            task_id='echo_task',
            bash_command=f'echo "Running {dag_id}"'
        )
        return dag

# 複数のDAGを自動生成（管理しやすくする）
for i in range(3):
    dag_id = f'generated_dag_{i}'
    globals()[dag_id] = generate_dag(dag_id)

from airflow.operators.python import BranchPythonOperator

# 条件分岐を決定する関数
def choose_branch():
    return 'task_a' if True else 'task_b'

# 分岐処理を行うオペレーター
branch_task = BranchPythonOperator(
    task_id='branch_task',
    python_callable=choose_branch
)

from airflow.sensors.filesystem import FileSensor

# 指定されたファイルが存在するか監視するセンサー
file_sensor = FileSensor(
    task_id='wait_for_file',
    filepath='/path/to/file',
    poke_interval=30,  # 30秒ごとにチェック
    timeout=600  # 10分間待機
)