Airflow Macro PythonOperator에 적용하기

Airflow Macro란?

지정된 변수를 호출해서 태스크에 적용할 수 있는 기능입니다.

만약 실행하는 코드에 날짜 설정이 now() 등으로 되어 있다면 backfill이나 과거의 작업을 재실행할 때 문제가 되는데,

이 때 {{execution_date}} 처럼 airflow task가 실행되는 날짜를 불러와서 적용할 수 있다는 것이 장점입니다.

 

airflow.apache.org/docs/apache-airflow/stable/macros-ref.html

Macros reference — Airflow Documentation

BashOperator에서는 매크로를 그대로 기입하면 돼서 어려움이 없습니다.

t2 = BashOperator(
    task_id='what_time',
    bash_command='echo {{ ds }}',
    dag=dag
)

 

다만 PythonOperator에서는 몇 가지 참고할 부분이 있는데요,

def what_time(**kwargs):
    ds = kwargs["ds"]
    print(ds)
    
    
t1 = PythonOperator(
    task_id='what_time',
    python_callable=what_time,
    provide_context=True,
    dag=dag
)

Python 코드에서 kwargs로 매크로를 지정해주는 것과,

Task 작성 시 provide_context = True를 꼭 설정해주어야 합니다.

 

이렇게 작성하고 task log를 보면 아래와 같이 ds로 입력한 부분이 execution_date로 치환되어 나타나는 것을 확인할 수 있습니다.