[Airflow] airflow decorators

[Airflow] airflow decorators

airflow decorators

airflow.decorators는 Airflow에서 제공하는 데코레이터 모듈

 

@dag

: DAG를 정의할 때 사용되는 데코레이터

DAG 객체를 생성하고, DAG 속성을 설정하는 데 사용됨 

 

@task

: Task를 정의할 때 사용되는 데코레이터

Python 함수를 task로 등록하고, task의 실행을 제어하는데 사용됨

 

@task_decorator

: 커스텀 task 데코레이터를 정의할 때 사용되는 데코레이터

이를 사용하여 task에 사용자 지정 로직이나 기능을 추가할 수 있음

 

@task_group

: task group을 정의할 때 사용되는 데코레이터

여러 task를 그룹으로 묶어서 실행 순서나 의존성을 관리할 수 있음

 

@task_trigger_dag

: 다른 DAG를 트리거할 때 사용되는 데코레이터

해당 task의 실행 후, 지정된 dag를 트리거 할 수 있음 

 

example

기존 방식

세 개의 task는 PythonOperator를 통해 생성됨 

from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago

def print_message(message):
    print(message)

default_args = {
    'start_date': days_ago(1),
}

dag = DAG(
    'my_dag',
    default_args=default_args,
    schedule_interval='@once',
    catchup=False
)

task1 = PythonOperator(
    task_id='task1',
    python_callable=print_message,
    op_args=['Task 1'],
    dag=dag
)

task2 = PythonOperator(
    task_id='task2',
    python_callable=print_message,
    op_args=['Task 2'],
    dag=dag
)

task3 = PythonOperator(
    task_id='task3',
    python_callable=print_message,
    op_args=['Task 3'],
    dag=dag
)

task1 >> task2 >> task3

데코레이터 모듈 사용 

@dag 데코레이터를 사용하여 DAG 객체를 생성하고,

@task 데코레이터를 사용하여 세 개의 테스크를 정의 

from datetime import datetime
from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

default_args = {
    'start_date': days_ago(1),
}

@dag(default_args=default_args, schedule_interval='@once', catchup=False)
def decorator_test_dag():
    
    @task
    def print_message(message):
        print(message)
    
    task1 = print_message('Task 1')
    task2 = print_message('Task 2')
    task3 = print_message('Task 3')

    task1 >> task2 >> task3

dag = decorator_test_dag()

 

✔ 데코레이터 모듈을 사용하여 Airflow DAG를 정의하면

간결하고 가독성이 높은 코드를 짤 수 있다