[Airflow]Jinja 템플릿

1. Jinja 템플릿

• 문서(파일)에서 특정 양식으로 작성된 값을 런타임시 실제 값으로 치환해주는 처리 엔진
• 템플릿 엔진은 여러 솔루션이 존재하며 그 중 Jinja 템플릿은 파이썬 언어에서 사용하는 엔진

 

• 파이썬 기반 웹 프레임워크인 Flask, Django에서 주로 사용됨
  (주로 HTML 템플릿 저장 후 화면에 보여질 때 실제 값으로 변환해서 출력)
• SQL 작성시에도 활용 가능

SELECT * FROM TABLES WHERE BASE_DT = {{ }}

 

 

 

 

2. Airflow 에서 사용법

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

https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html

Templates reference — Airflow Documentation

 

Variable	Type	Description
{{ date_interval_start }}	pendulum.DateTime	Start of the data interval
{{ date_interval_end }}	pendulum.DateTime	End of the data interval
{{ ds }}	str	The DAG run's logical date as YYYY-MM-DD Same as {{ dag_run.logical_date | ds }}
{{ ds_nodash }}	str	Same as {{ dag_run.logical_date | ds_nodash }}
{{ ts }}	str	Same as {{ dag_run.logical_date | ts }} Example : 2018-01-01T00:00:00+00:00
{{ ts_nodash }}	str	Same as {{ dag_run.logical_date | ts_nodash }} Example: 20180101T000000
...

 

• 모든 오퍼레이터, 모든 파라미터에 Template 변수를 적용할 수는 없음
  Airflow 문서에서 어떤 파라미터에 Template 변수 적용 가능한지 확인 필요함
• 아래 링크에서 확인할 수 있음

https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/index.html

airflow.operators — Airflow Documentation

 

• Python 오퍼레이터에서는 op_kwargs, op_args, templates_dict 등에서 Template 변수가 사용가능함

 

 

 

3. Bash 오퍼레이터 Template

• Bash 오퍼레이터는 어떤 파라미터에 Template를 쓸 수 있는가?
• 파라미터
  • bash_command (str)
  • env (dict[str, str] | None)
  • append_env (bool)
  • output_encoding (str)
  • skip_exit_code (int)
  • cwd (str | None)

4. Bash 오퍼레이터 Template 사용 예제코드

from __future__ import annotations

import datetime

import pendulum   # datetime을 좀더 편하게 사용할 수 있게해주는 라이브러리

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator


with DAG(
    dag_id = "dags_bash_with_template",
    schedule = "10 0 * * *",
    start_date = pendulum.datetime(2023, 3, 1, tz="Asia/Seoul"),
    catchup = False
) as dag:
    bash_t1 = BashOperator(
        task_id = 'bash_t1',
        bash_command = 'echo "data_interval_start: {{ data_interval_start }}" && echo "data_interval_end: {{ data_interval_end }}"'
    )

    bash_t2 = BashOperator(
        task_id = 'bash_t2',
        env = {'START_DATE': '{{ data_interval_start | ds }}',
            'END_DATE':'{{ data_interval_end | ds }}'
        },
        bash_command = 'echo $START_DATE && echo $END_DATE'
    )

    bash_t1 >> bash_t2

 

• data_interval_start와 data_interval_end의 차이

속성	data_interval_start	data_interval_end
의미	데이터 인터벌의 시작 시점	데이터 인터벌의 끝 시점
타임스템프예시	2023-01-09 00:00:00	2023-01-10 00:00:00
활용시점	데이터 처리를 시작하는 기준 시점	데이터 처리를 끝내는 기준 시점
예시	만약 DAG이 매일 00:00에 실행되도록 스케줄링되었다면, 이 값은 현재 실행 주기의 시작인 전날 00:00을 가리킵니다.	동일한 DAG이 매일 00:00에 실행된다면, 이 값은 현재 실행 주기의 끝인 당일 00:00을 가리킵니다.

 

 

 

bash_t1 실행결과

bash_t2 실행결과