[Airflow] Python 오퍼레이터 macro 사용하기

1. Python 오퍼레이터 with macro

• Template 변수를 지원하는 파라미터
  • python_callable (Callable | None)
  • op_kwargs
  • op_args
  • templates_dict
  • temlates_exts
  • show_return_value_in_logs

 

2. Python 오퍼레이터 with macro 사용 예제 코드

/*dags_python_with_macro.py*/

from airflow import DAG
import pendulum
from airflow.decorators import task

with DAG(
    dag_id="dags_python_with_macro",
    schedule="10 0 * * *",
    start_date=pendulum.datetime(2023, 3, 1, tz="Asia/Seoul"),
    catchup=False
) as dag:

    # Macro 사용 예제
    @task(task_id='task_using_macros',
        templates_dict={
            'start_date':'{{ (data_interval_end.in_timezone("Asia/Seoul") + macros.dateutil.relativedelta.relativedelta(months=-1, day=1)) | ds }}' # 한달전 1일
            ,'end_date': '{{(data_interval_end.in_timezone("Asia/Seoul").replace(day=1) + macros.dateutil.relativedelta.relativedelta(days=-1)) | ds }}' # 일자를 1일로 바꾸고 하루를 더 빼서 전월 마지막일이 나옴
        }
    )
    def get_datetime_macro(**kwargs):
        templates_dict = kwargs.get('templates_dict') or {}
        if templates_dict:
            start_date = templates_dict.get('start_date') or 'start_date없음'
            end_date = templates_dict.get('end_date') or 'end_date없음'
            print(start_date)
            print(end_date)

    # 직접 연산 예제
    @task(task_id='task_direct_calc')
    def get_datetime_calc(**kwargs):
        from dateutil.relativedelta import relativedelta    # 여기에서 import 하면 이부분은 스케줄러가 검사하지 않아서 부하를 줄일 수 있음
        data_interval_end = kwargs['data_interval_end']
        # 직접 연산하는 부분
        prev_month_day_first = data_interval_end.in_timezone('Asia/Seoul') + relativedelta(months=-1, day=1)
        prev_month_day_last = data_interval_end.in_timezone('Asia/Seoul').replace(day=1) + relativedelta(days=-1)
        print(prev_month_day_first.strftime('%Y-%m-%d'))
        print(prev_month_day_last.strftime('%Y-%m-%d'))


    get_datetime_macro() >> get_datetime_calc()

 

get_datetime_macro() (Macro 사용 예제) 실행결과

 

get_datetime_calc() (직접 연산 예제) 실행결과