Airflow | Airflow variable, airflow variable 사용법, variables를 사용한 전역변수 설정

 

airflow를 사용하다가 DAG 코드와 설정값을 어떻게 분리해야 할지 고민이 될 때가 있었다.

이런 고민을 해결하기 위해 airflow는 Variable이라는 기능을 제공하고 있다.

https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/variables.html

Variables — Airflow 3.1.6 Documentation

하지만 Variable은 잘 쓰면 약, 잘못 쓰면 장애의 원인이 되기도 하는 만큼 개념과 역할, 올바른 사용법 등을 숙지하고 있어야 한다.

 

 

1. Airflow에서 Variable이란

airflow variable은 airflow metadata DB에 저장되는 전역 key-value 설정 값이다.

Variable은 DAG 코드와 분리되어 실행 시점에 동적으로 조회되는 설정 값이다.

 

S3_BUCKET = "prod-raw-data-bucket"

이렇게 DAG 코드에 하드코딩을 하게 되면 환경 분리가 어려우며 설정 변경 시 DAG 코드 수정이 필요하다는 문제가 있다.

 

variable을 사용하면

from airflow.models import Variable

S3_BUCKET = Variable.get("S3_BUCKET")

위와 같이 DAG 코드에서 사용이 가능하다. 이는 코드 수정없이 설정 변경이 가능하며 운영 중에도 즉시 반영이 가능하다. 또 환경 별로 설정 관리도 가능하다는 장점이 있다.

 

 

2. Airflow Variable 생성 방법

airflow web UI에서 admin -> variables -> add 순서대로 페이지 접속을 하면 variable울 추가하고 수정, 삭제할 수 있다.

ex) Key - S3_BUCKET  / Value - prod-raw-data-bucket

 

JSON 값으로도 저장이 가능하며 이는 즉시 반영 된다. key에 variables 이름을 value에 아래와 같이 여러 valriable을 json 값으로 넣어주면 된다.

key - Test Variables

Values -

{
  "S3_BUCKET": "prod-raw-data-bucket",
  "RETRY_COUNT": 3
}

 

CLI로 추가하는 방법도 있다. airflow 환경 터미널에서

airflow variables set S3_BUCKET prod-raw-data-bucket
airflow variables get S3_BUCKET

이렇게 입력해 주면 web UI에서의 key, value 값으로 각각 설정된다.

 

 

3.  DAG에서 Variable 사용법

from airflow.models import Variable

bucket = Variable.get("S3_BUCKET")

DAG 파일에서 Variable을 import 해 주면 get 함수로 variable들을 가져올 수 있고 variable은 항상 문자열로 반환된다.

Variable은 task 실행 시점에 읽어오기 때문에 아래처럼 사용하는 게 권장된다.

def build_env_vars():
    return {
        "SRC_USER": Variable.get("S3_BUCKET"),
        "SRC_HOST": Variable.get("RETRY_COUNT"),
    }
    

DockerOperator(
    task_id="Test_Variables",
    ...
    environment=build_env_vars(),
)

이렇게 사용하게 되면 DAG 파싱 단계에서 DB 접근이 없으며 실행 시점 기준 설정을 사용할 수 있다.

 

이때 dockeroperator 사용시 environment는 추가가 아니라 덮어쓰기 개념이기 때문에 dockerfile의 PYTHONPATH와 같은 환경 변수가 사라질 수 있다.

때문에 dockeroperator 사용 시 기존 환경에 variable을 병합하는 형태로 사용해야 한다.

import os
from airflow.models import Variable

def build_env_vars():
    base_env = dict(os.environ)

    pipeline_env = Variable.get(
        "Test_Varialbes",
        deserialize_json=True
    )

    base_env.update(Test_Variables)
    return base_env

 

 

4. Variable에 넣으면 안되는 것들

DB 비밀번호, API Token, Access Key와 같은 값들은 보안이 필요하기 때문에 variable에 넣는 것이 권장되지 않는다.

variable은 설정 값을 위한 도구이지 보안 정보 저장소가 아니다.