[Airflow] SubDag 개념, 장단점, 샘플 코드 (feat. ChatGPT)

SubDag

- SubDag는 Airflow DAG 안에 존재하는 작은 DAG

- 이를 사용하면 하나의 큰 DAG를 여러 작은 DAG로 분리 가능

- 이렇게 작은 DAG로 분리함으로써, 개발, 테스트 및 유지보수가 더 쉬움

- 또한, SubDag는 DAG를 이해하기 쉬운 작은 단위로 나눌 수 있으며, 작은 단위별로 추적할 수 있어 일부 작업이 실패한 경우 재시작 가능

- SubDag는 DAG 생성 시 하위 레벨의 DagBag에 의해 생성

- 하위 레벨의 DAG는 SubDagOperator를 사용하여 DAG 내에서 호출

 

SubDag의 장단점

장점

1. 큰 DAG를 여러 작은 DAG로 분리할 수 있으므로 개발, 테스트 및 유지보수가 더 쉬워집니다.
2. 작은 DAG로 나누어짐으로써 DAG를 이해하기 쉬운 작은 단위로 나눌 수 있습니다.
3. 작은 단위별로 추적할 수 있어 일부 작업이 실패한 경우 재시작할 수 있습니다.

단점

1. SubDag가 많아지면 DAG 템플릿의 복잡성이 증가합니다.

샘플 코드

아래는 SubDag를 사용한 예제 코드입니다.

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.subdag_operator import SubDagOperator

from subdags.my_subdag import subdag

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2022, 3, 12),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    dag_id='my_dag',
    default_args=default_args,
    schedule_interval=timedelta(days=1),
)

def print_hello():
    return 'Hello!'

def print_world():
    return 'World!'

with dag:
    hello_task = PythonOperator(
        task_id='hello_task',
        python_callable=print_hello,
    )

    world_task = PythonOperator(
        task_id='world_task',
        python_callable=print_world,
    )

    with SubDagOperator(
        task_id='subdag_task',
        subdag=subdag('my_dag', 'subdag_task', default_args),
        dag=dag,
    ) as subdag_task:
        sub_task_1 = PythonOperator(
            task_id='sub_task_1',
            python_callable=print_hello,
            dag=subdag_task,
        )

        sub_task_2 = PythonOperator(
            task_id='sub_task_2',
            python_callable=print_world