[Airflow] Jinja Template with Airflow

Jinja Template이란?

•

Jinja 템플릿은 Python에서 널리 사용되는 템플릿 엔진

◦

Django 템플릿 엔진에서 영감을 받아 개발

◦

Jinja를 사용하면 프레젠테이션 로직과 애플리케이션 로직을 분리하여 동적으로 HTML 생성

◦

Flask에서 사용됨

▪

변수는 이중 중괄호 {{ }}로 감싸서 사용

<h1>안녕하세요, {{ name }}님!</h1>
Python
복사

▪

제어문은 퍼센트 기호 {% %}로 표시

<ul>
{% for item in items %}
 <li>{{ item }}</li>
{% endfor %}
</ul>
Python
복사

참고: Jinja Template + Airflow

•

Airflow에서 Jinja 템플릿을 사용하면 작업 이름, 파라미터 또는 SQL 쿼리와 같은 작업 매개변수를 템플릿화된 문자열로 정의 가능

◦

이를 통해 재사용가능하고 사용자 정의 가능한 워크플로우 생성

•

예 1) execution_date 과 같은 파라미터를 코드 내에서 쉽게 사용: {{ ds }}

◦

가능한 모든 시스템 변수는 여기를 참조

Templates reference — Airflow Documentation

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

# BashOperator를 사용하여 템플릿 작업 정의
task1 = BashOperator(
 task_id='task1',
 bash_command='echo "{{ ds }}"',
 dag=dag
)
Python
복사

•

예 2) 파라미터 등으로 넘어온 변수를 쉽게 사용 가능

# 동적 매개변수가 있는 다른 템플릿 작업 정의
task2 = BashOperator(
 task_id='task2',
 bash_command='echo "안녕하세요, {{ params.name }}!"',
 params={'name': 'John'}, # 사용자 정의 가능한 매개변수
 dag=dag
)
Python
복사

참고: BashOperator 레퍼런스 보기

airflow.operators.bash — Airflow Documentation

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

•

bash_command (str) – The command, set of commands or reference to a bash script (must be ‘.sh’) to be executed. (templated)

•

env (dict[str, str] | None) – If env is not None, it must be a dict that defines the environment variables for the new process; these are used instead of inheriting the current process environment, which is the default behavior. (templated)

•

append_env (bool) – If False(default) uses the environment variables passed in env params and does not inherit the current process environment. If True, inherits the environment variables from current passes and then environment variable passed by the user will either update the existing inherited environment variables or the new variables gets appended to it

•

output_encoding (str) – Output encoding of bash command

•

skip_on_exit_code (int | Container[int] | None) – If task exits with this exit code, leave the task in skipped state (default: 99). If set to None, any non-zero exit code will be treated as a failure.

•

cwd (str | None) – Working directory to execute the command in. If None (default), the command is run in a temporary directory

참고: Airflow에서 사용 가능한 Jinja 변수들 몇개 살펴보기

•

{{ ds }} 

◦

현재 실행 중인 DAG 인스턴스의 실행 날짜 (날짜 문자열 형식: "YYYY-MM-DD").

•

{{ ds_nodash }}

◦

현재 실행 중인 DAG 인스턴스의 실행 날짜 (날짜 문자열 형식: "YYYYMMDD").

•

{{ ts }}

◦

현재 실행 중인 태스크의 시작 시간 (시간 및 날짜 문자열 형식: "YYYY-MM-DDTHH:MM:SS")

•

{{ dag }}

◦

현재 실행 중인 DAG 인스턴스의 DAG 객체.

•

{{ task }}

◦

현재 실행 중인 태스크 객체.

•

{{ dag_run }}

◦

현재 실행 중인 DAG 인스턴스의 DAGRun 객체.

•

{{ var.value }}

◦

{{ var.value.get('my.var', 'fallback') }}

◦

Airflow 변수의 값을 가져옵니다. 

◦

var은 Airflow에서 정의된 변수의 이름을 나타내며, value는 해당 변수의 값을 가져옵니다. 

◦

예를 들어, {{ var.value.get('my.var', 'fallback') }}는 my.var라는 변수의 값을 가져오며, 값이 없는 경우 'fallback'을 반환합니다.

•

{{ var.json }}

◦

{{ var.json.my_dict_var.key1 }}

◦

Airflow 변수의 JSON 형식 값을 가져옵니다. 

◦

var.json은 JSON 형식의 변수 값을 가져오며, 이를 통해 내부 속성에 접근할 수 있습니다. 

◦

예를 들어, {{ var.json.my_dict_var.key1 }}는 my_dict_var라는 변수의 key1 속성 값을 가져옵니다.

•

{{ conn }}

◦

{{ conn.my_conn_id.login }}, {{ conn.my_conn_id.password }}

◦

Airflow 연결 (Connection)의 속성에 접근합니다. 

◦

conn은 Airflow에서 정의된 연결의 이름을 나타내며, my_conn_id라는 연결의 login 및 password 속성 값에 접근하는 예시는 아래와 같습니다

◦

{{ conn.my_conn_id.login }}, {{ conn.my_conn_id.password }}.

참고: Airflow에서 Jinja 변수를 사용한 예제 코드 살펴보기

•

Learn_Jinja

learn-airflow/dags/Learn_Jinja.py at main · learndataeng/learn-airflow

Contribute to learndataeng/learn-airflow development by creating an account on GitHub.

https://github.com/learndataeng/learn-airflow/blob/main/dags/Learn_Jinja.py

◦

BashOperator 3개로 구성

▪

“airflow dags test Learn_Jinja 2023-05-30”

•

코드

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

# DAG 정의
dag = DAG(
    'Learn_Jinja',
    schedule='0 0 * * *',  # 매일 실행
    start_date=datetime(2023, 6, 1),
    catchup=False
)

# BashOperator를 사용하여 템플릿 작업 정의
task1 = BashOperator(
    task_id='task1',
    bash_command='echo "{{ ds }}"',
    dag=dag
)

# 동적 매개변수가 있는 다른 템플릿 작업 정의
task2 = BashOperator(
    task_id='task2',
    bash_command='echo "안녕하세요, {{ params.name }}!"',
    params={'name': 'John'},  # 사용자 정의 가능한 매개변수
    dag=dag
)

task3 = BashOperator(
    task_id='task3',
    bash_command="""echo "{{ dag }}, {{ task }}, {{ var.value.get('csv_url') }}" """,
    dag=dag
)

task1 >> task2 >> task3
Python
복사