[Airflow] Bash Operator 와 macros

1. Macro 변수 이해

• Macro 변수의 필요성

sql = f'''
SELECT NAME, ADDRESS
FROM TBL_REG
WHERE REG_DATE BETWEEN ?? AND ??
'''

• 위와 같은 쿼리가 있을 때 REG_DATE의 범위를 복잡하게 설정하려면 Macro 변수가 필요함
  ex)
  - 배치일이 1월31일이라면 12월31일부터 1월30일까지
  - 배치일이 2월28일이라면 1월31일부터 2월27일까지
• Template 변수 기반 다양한 날짜 연산이 가능하도록 연산 모듈을 제공하고 있음
  ex) macros.datetime, macros.dateutil
• Macro를 잘 쓰려면 파이썬 datetime 및 dateuil 라이브러리에 익숙해져야 함

 

2. 파이썬의 datetime, dateutil 라이브러리 사용 예제코드

from datetime import datetime
from dateutil import relativedelta

now = datetime(year=2023, month=3, day=30)
print('현재시간:'+str(now))
print('---------------월 연산---------------')
print(now + relativedelta.relativedelta(month=1)) # 1월로 변경
print(now.replace(month=1)) # 1월로 변경
print(now + relativedelta.relativedelta(months=-1)) # 1개월 빼기
print('---------------일 연산---------------')
print(now + relativedelta.relativedelta(day=1)) # 1일로 변경
print(now.replace(day=1)) # 1일로 변경
print(now + relativedelta.relativedelta(days=-1)) # 1일 빼기
print('---------------연산 여러개---------------')
print(now + relativedelta.relativedelta(months=-1) + relativedelta.relativedelta(days=-1)) # 1개월, 1일 빼기기

실행결과

 

 

3. Bash 오퍼레이터 with macro 실습해보기

• 예시1) 매월 말일 수행되는 Dag에서
  변수 START_DATE : 전월말일,
  변수 END_DATE : 어제로 env 셋팅하기
  
  

"""
dags_bash_with_macro_eg1.py
"""

from airflow import DAG
import pendulum
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="dags_bash_with_macro_eg1",
    schedule="10 0 L * *",
    start_date=pendulum.datetime(2023, 3, 1, tz='Asia/Seoul'),
    catchup=False
) as dag:
    # START_DATE : 전월 말일, END_DATE : 1일 전
    bash_task_1 = BashOperator(
        task_id = 'bash_task_1',
        env={
            'START_DATE':'{{data_interval_start.in_timezone("Asia/Seoul") | ds}}',
            'END_DATE':'{{(data_interval_end.in_timezone("Asia/Seoul") - macros.dateutil.relativedelta.relativedelta(days=1)) | ds}}'
        },
        bash_command = 'echo "START_DATE: $START_DATE" && echo "END_DATE: $END_DATE"'
    )

실행결과

 

 

 

• 예시2) 매월 둘째주 토요일에 수행되는 Dag에서
  변수 START_DATE : 2주 전 월요일
  변수 END_DATE : 2주 전 토요일로 env 셋팅하기
  
  

"""
dags_bash_with_macro_eg2.py
"""

from airflow import DAG
import pendulum
from airflow.operators.bash import BashOperator

with DAG(
    dag_id="dags_bash_with_macro_eg2",
    schedule="10 0 * * 6#2",
    start_date=pendulum.datetime(2023, 3, 1, tz='Asia/Seoul'),
    catchup=False
) as dag:
    # START_DATE : 2주전 월요일, END_DATE : 2주전 토요일
    bash_task_2 = BashOperator(
        task_id = 'bash_task_2',
        env={
            'START_DATE':'{{(data_interval_end.in_timezone("Asia/Seoul") - macros.dateutil.relativedelta.relativedelta(days=19)) | ds}}',
            'END_DATE':'{{(data_interval_end.in_timezone("Asia/Seoul") - macros.dateutil.relativedelta.relativedelta(days=14)) | ds}}'
        },
        bash_command = 'echo "START_DATE: $START_DATE" && echo "END_DATE: $END_DATE"'
    )

 

실행결과