Variable

상위: Airflow

요약

Variable은 Airflow에서 여러 DAG 및 Task 간 설정값을 공유하기 위한 전역 변수입니다. 메타 DB에 영구 저장되며, Web UI, CLI, API를 통해 관리할 수 있습니다. XCom과 달리 모든 DAG에서 접근 가능하며, 환경별 설정 관리에 유용합니다.

Variable이란?

• Airflow에서 여러 DAG 및 Task 간 데이터 공유를 위한 변수
• 모든 DAG에서 공유 가능
• 협업 시 표준화된 DAG 구성에 사용
• 메타 DB에 영구 저장
• Airflow UI, CLI, API를 통해 관리 가능

Variable vs XCom

Variable 등록

Web UI에서 등록

1. Airflow UI 접속
2. Admin → Variables 클릭
3. [+] 버튼 클릭

4. Key / Value / Description 입력 후 Save

CLI로 등록

# Variable 설정
airflow variables set my_key my_value

# JSON 형식으로 설정
airflow variables set my_json_key '{"name": "John", "age": 30}'

Python 코드로 설정

from airflow.models import Variable

# Variable 설정
Variable.set("my_key", "my_value")

# JSON 설정
Variable.set("config", {"db_host": "localhost", "db_port": 5432}, serialize_json=True)

Variable 조회

기본 조회

from airflow.models import Variable

def use_variable():
    my_var = Variable.get("my_key")
    print(f"Variable 값: {my_var}")

기본값 지정

# Variable이 없을 경우 기본값 반환
my_var = Variable.get("my_key", default_var="default_value")

JSON 조회

# JSON으로 저장된 Variable 조회
config = Variable.get("config", deserialize_json=True)
db_host = config['db_host']
db_port = config['db_port']

사용 예제

환경별 설정

def get_db_config():
    env = Variable.get("environment", default_var="dev")
    
    if env == "prod":
        db_host = Variable.get("prod_db_host")
    else:
        db_host = Variable.get("dev_db_host")
    
    return db_host

DAG에서 활용

from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator

def process_data(**context):
    api_key = Variable.get("api_key")
    threshold = int(Variable.get("threshold", default_var="100"))
    
    # API 호출 로직
    print(f"Using API key: {api_key[:5]}...")
    print(f"Threshold: {threshold}")

with DAG('example_dag', ...) as dag:
    task = PythonOperator(
        task_id='process',
        python_callable=process_data
    )

주의사항

DAG 파싱 시 Variable.get() 금지

# ❌ 잘못된 사용 (DAG 파싱 시마다 DB 접근)
my_var = Variable.get("my_key")

dag = DAG(
    dag_id='bad_example',
    default_args={'param': my_var},  # 파싱 시마다 실행
    ...
)

# ✅ 올바른 사용 (Task 실행 시에만 접근)
def get_variable():
    my_var = Variable.get("my_key")
    # 작업 수행

dag = DAG(dag_id='good_example', ...)
task = PythonOperator(
    task_id='task',
    python_callable=get_variable  # Task 실행 시에만 Variable 조회
)

민감 정보 저장

# ⚠️ Variable에 민감 정보 저장 시 주의
# 민감한 정보는 Connections 사용 권장

# ❌ Variable에 비밀번호 저장
Variable.set("db_password", "secret123")

# ✅ Connection 사용
from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection('my_db_conn')
password = conn.password

자세한 내용은 Connections & Hooks 참조

성능 고려

• Variable.get()은 DB 조회 발생
• 반복 호출 시 성능 저하 가능
• 한 번 조회 후 로컬 변수에 저장 권장

# ❌ 반복 호출
for i in range(100):
    value = Variable.get("my_key")  # 100번 DB 조회

# ✅ 한 번만 조회
value = Variable.get("my_key")
for i in range(100):
    # value 사용

Variable 삭제

CLI

airflow variables delete my_key

Python

from airflow.models import Variable

Variable.delete("my_key")

실전 패턴

환경별 설정 관리

# Variables 설정
# environment: "dev" | "staging" | "prod"
# dev_db_url: "postgresql://dev..."
# staging_db_url: "postgresql://staging..."
# prod_db_url: "postgresql://prod..."

def get_db_url():
    env = Variable.get("environment")
    db_url = Variable.get(f"{env}_db_url")
    return db_url

Feature Flag

def conditional_task(**context):
    feature_enabled = Variable.get("new_feature_enabled", default_var="false")
    
    if feature_enabled == "true":
        # 새 기능 실행
        print("New feature enabled")
    else:
        # 기존 기능 실행
        print("Using legacy feature")

다음 단계

• XCom - Task 간 데이터 전달
• Connections & Hooks - 외부 시스템 연결
• DAG - DAG 기본 개념

참고 자료

• Variable 공식 문서
• Variable API 참조