예제

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

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

    @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 }}',
            "end_date": '{{ data_interval_end.in_timezone("Asia/Seoul").replace(day=1) + macros.dateutil.relativedelta.relativedelta(days=-1) | ds}}',
        },
    )
    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)

    get_datetime_macro()

하지만 굳이 Python Operator에서 Macro를 사용할 필요가 있을까?

	@task(task_id="task_direct_calc")
    def get_datetime_calc(**kwargs):
        from dateutil.relativedelta import relativedelta

        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"))

여기서 from, import 자체를 task 내에서 선언한 것을 볼 수 있는데, 이는 스케줄러 부하 경감을 위해서입니다.

스케줄러는 주기적으로 만든 DAG를 Parsing 하는데, import 하는 부분, with dag 시작하기 전 부분, operator 선언 전 부분에 내용이 많을 수록 스케줄러는 부하를 많이 받습니다.

대규모 환경에서는 스케줄러의 부하를 줄이는 부분이 중요한데, 그래서 가급적이면 스케줄러 부하를 줄이기 위해서 오퍼레이터 안에서만 선언하는 것이 하나의 팁입니다.

중요한 부분은 Python Operator 사용 시에, macro로 날짜를 연산할 것인지, 밖에서 연산을 할 것인지는 편한 방향으로 진행하면 됩니다.