[Airflow] 전역 공유변수 Variable

1. 전역변수 Variable 이해

• Xcom: 특정 DAG, 특정 schedule에 수행되는 Task 간에만 공유
• 모든 DAG이 공유할 수 있는 전역변수
• 전역변수를 사용하는 경우
  • 현업 환경에서 표준화된 dag을 만들기 위해 주로 사용
  • 주로 상수(CONST)로 지정해서 사용할 변수들 셋팅
    ex.
    bash_sh_dir=/opt/airflow/plugins/shell
    bash_file_dir=/opt/airflow/plugins/files
    email, Alert 메시지를 받을 담당자의 email 주소 정보 등
• Variable 등록하기
  (1) Admin -> Variables 로 이동
  (2) [+] 버튼 클릭
  (3) Add Variable 창에서 각 값을 입력 후 save버튼 클릭

 

 

새로운 변수가 등록된 모습

 

 

 

 

2. 전역변수 사용하기

• 1안 :Variable 라이브러리 이용
  파이썬 문법을 이용해 미리 가져오기
  (스케줄러의 주기적 DAG 파싱 시 Variable.get 개수만큼 DB연결을 일으켜 불필요한 부하 발생)
  (스케줄러 과부화 원인 중 하나)

from airflow.operators.bash import BashOperator
from airflow.models import Variable

var_value = Variable.get("sample_kye")
bash_var1 = BashOperator(
	task_id = "bash_var_1",
    bash_command = f"echo variable:{var_value}"
)

 

• 2안 :Jinja 템플릿 이용 ( 해당 방법 권고 )
  오퍼레이터 내부에서 가져오기

from airflow.operators.bash import BashOperator

bash_var_2 = BashOperator(
	task_id = "bash_var_2",
    bash_command = "echo variable:{{var.value.sample_key}}"
)

 

 

 

2. 전역변수 사용하기 예제코드

"""
dags_bash_with_variable.py
"""

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


with DAG(
    dag_id = 'dags_bash_with_variable',
    schedule = "10 9 * * *",
    start_date = pendulum.datetime(2023, 4, 1, tz="Asia/Seoul"),
    catchup = False
) as dag:
    # 첫 번째 방법 (사용 x 권고)
    var_value = Variable.get("sample_key")

    bash_var_1 = BashOperator(
        task_id = "bash_var_1",
        bash_command = f"echo variable:{var_value}"
    )

    # 두 번째 방법 (사용 o 권고)
    bash_var_2 = BashOperator(
        task_id = "bash_var_2",
        bash_command = "echo variable:{{var.value.sample_key}}"
    )

 

bash_var1 실행결과

 

 

bash_var2 실행결과