[Python/Jinja/Airflow] Jinja 템플릿과 Airflow에서의 사용방법

Jinja 템플릿

 

Jinja 템플릿은 파이썬 언어에서 사용하는 템플릿 엔진으로, 문서(파일)에서 특정 양식으로 작성된 값을 런타임시 실제 값으로 치환해주는 처리 엔진이다.

 

Jinja 템플릿은 주로 파이썬 기반 웹 프레임워크인 Flask, Django, FastAPI에서 주로 사용한다. 이 경우 html 템플릿에 내용을 저장하고 화면에 보여질 때 실제 값으로 변환해서 출력한다. 각 상황에 맞게 변환해서 html 템플릿을 보여줄 수 있기 때문에, html 파일의 재활용성이 높아져 효율적으로 이용할 수 있다.

 

SQL 작성 시에도 Jinja 템플릿을 활용할 수 있다. (상황에 따라 {{ }}의 값만 바뀌도록 적용 가능)

ex) select * from tables where base_dt = {{ }}

 

 

Airflow에서의 Jinja 템플릿 활용

 

Airflow에서는 오퍼레이터 파라미터 입력 시 중괄호 2개 ( {{ }} )를 이용하면 Airflow에서 기본 제공하는 변수들을 치환된 값으로 입력할 수 있다. (ex. 수행 날짜, DAG_ID 등)

공식문서: https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html

Templates reference — Airflow Documentation

 

그러나 모든 오퍼레이터와 파라미터에서 템플릿을 이용할 수 있는 것은 아니다.

Airflow 공식 문서 내 Operators 부분에서 Parameters 각 줄 뒤에 (templated)와 같은 스트링이 붙어있어야 템플릿을 활용할 수 있다. 그 외에는 활용할 수 없다.

 

정확히는 좀 더 하단에 있는 이 부분을 보면 어떤 파라미터가 템플릿을 활용할 수 있는지 알 수 있다.

 

 

Bash Operator에서의 Jinja 템플릿 사용

 

Bash 오퍼레이터에서는 아래 파라미터에서 템플릿을 사용할 수 있다.

• bash_command (str)
• env (dict[str, str] | None)

템플릿의 기능 중 날짜를 Bash 오퍼레이터를 통해 출력해보겠다.

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

with DAG(
    dag_id="dags_bash_select_fruit",
    schedule="10 0 * * 6#1", # 첫 번째 토요일 0시 10분마다
    start_date=pendulum.datetime(2023, 3, 1, tz="Asia/Seoul"),
    catchup=False,
    tags=["practice"]
) as dag:
    
    bash_t1 = BashOperator(
        task_id = 'bash_t1',
        bash_command = 'echo "end date is {{ data_interval_end }}"' #pendulum.datetime 형식으로 출력
    )

    bash_t2 = BashOperator(
        task_id = 'bash_t2',
        env = {'START_DATE': '{{ data_interval_start | ds}}', # 일종의 변수 정의 개념
               'END_DATE': '{{ data_interval_end | ds}}'}, # YYYY-MM-DD 형식으로 출력됨
        bash_command = 'echo "Start date is $START_DATE " &&'
                        'echo "End date is $END_DATE"'
    )

    bash_t1 >> bash_t2

 위 코드를 서버로 보낸 뒤, 로그를 확인해보면 아래와 같이 노출된다.

bash_t2

 

 

Python Operator에서의 Jinja 템플릿 사용

Python 오퍼레이터에서는 아래 파라미터에서 템플릿을 사용할 수 있다.

• templates_dict
• op_args
• op_kwargs

위 파라미터 중 op_kwargs를 이용해 템플릿을 사용해보려고 한다.

from airflow import DAG
import pendulum
import datetime
from airflow.operators.python import PythonOperator
from airflow.decorators import task

with DAG(
    dag_id="dags_python_template",
    schedule="30 9 * * *",
    start_date=pendulum.datetime(2023, 3, 10, tz="Asia/Seoul"),
    tags=["practice"],
    catchup=False
) as dag:
    
    def python_function1(start_date, end_date, **kwargs):
        print(start_date)
        print(end_date)

    python_t1 = PythonOperator( # 첫 번째 방법: templates 변수를 직접 부르는 방법
        task_id='python_t1',
        python_callable=python_function1,
        op_kwargs={'start_date':'{{data_interval_start | ds}}', 'end_date':'{{data_interval_end | ds}}'}
    )

    @task(task_id='python_t2')
    def python_function2(**kwargs): # 두 번째 방법: kwargs 딕셔너리 내 자동 제공되는 templates 변수들을 불러오기
        print(kwargs)
        print('ds:' + kwargs['ds'])
        print('ts:' + kwargs['ts'])
        print('data_interval_start:' + str(kwargs['data_interval_start']))
        print('data_interval_end:' + str(kwargs['data_interval_end']))
        print('task_instance:' + str(kwargs['ti']))


    python_t1 >> python_function2()

Python 오퍼레이터에서 templates 변수를 사용하는 방법은 두 가지가 있다.

 

1. templates 변수를 직접 사용하는 방법

- {{data_interval_start | ds}}와 같이 템플릿 변수를 직접 중괄호 내에 작성해 호출하는 방법이 있다.

 

2. **kwargs에 자동으로 제공되는 templates 변수들을 불러와 사용하는 방법

- 파이썬 오퍼레이터는 **kwargs에 template 변수를 자동으로 제공해준다. 딕셔너리 key값을 이용해 템플릿 값들을 사용할 수 있다.

 

 

위 두 방법 모두 위와 같이 정상적으로 출력되는 것을 확인해볼 수 있다. 개인적으로는 **kwargs를 이용한 방법이 좀 더 편하다고 생각되는데, 각자 원하는 방법대로 템플릿 변수를 사용하면 될 것 같다.