airflow Variable

1. Variables 개요

• airflow에서 사용하는 key-value 저장소의 역할을 한다.
• airflow에서 관리하는 글로벌 환경변수로 여러 DAG에서 공유해야 하는 설정 값을 저장한다.
• DAG 코드의 변경 없이 런타임에 설정을 수정할 수 있다.
• airflow 메타 데이터베이스에 저장되기에 너무 많은 Variables는 성능에 영향을 줄 수 있다.

 

2. Variables과 Fernet

airflow는 메타스토어 데이터베이스에 저장된 변수를 암호화하기 위해 Fernet을 사용한다.

airflow 설치 시 설정한 fernetKey가 사용되며, fernetKey를 설정하지 않고 airflow를 설치한 경우에는 암호화 되지 않는다.

UI 상에서도 is Encrypted의 값이 다른 것을 확인할 수 있다.

[ fernetKey를 설정하지 않은 airflow의 Variable UI ]

 

[ fernetKey를 설정한 airflow의 Variable UI ]

 

airflow의 메타 데이터베이스를 확인해보면 더 명확한 차이를 알 수 있다.

[ fernetKey를 설정하지 않은 airflow의 meta database ]

fernetKey를 설정하지 않아 value가 문자열로 저장된다.

bash-4.2$ psql -U airflow_user -d airflow_db -c "select * from variable"
could not change directory to "/home/airflow": Permission denied
 id | key  | val  | description | is_encrypted 
----+------+------+-------------+--------------
  1 | test | test |             | f

 

[ fernetKey를 설정한 airflow의 meta database ]

value가 fernetKey로 암호화되어 저장된다.

I have no name!@airflow-postgresql-0:/$ psql -U postgres -d postgres -c "select * from variable"
Password for user postgres:
 id | key  |                                                 val                                                  | description | is_encrypted
----+------+------------------------------------------------------------------------------------------------------+-------------+--------------
  1 | test | gAAAAABm1Yv7tyENd3Kbn9E7qPi-YeZUA6SthfP01fUzE0VVT1I_O6z_UUA5Mi0N2Emue6FDboEPEy3AjyvU1iZT8LMCE5UTPg== |             | t

 

 

아래와 같이 설정한 fernetKey를 통해 복호화하면, 본래의 문자열을 확인할 수 있다.

>>> from cryptography.fernet import Fernet
>>> f = Fernet("FDLyJe1N004_S5eTyZswSKqKm361wDFasO6l8q0mvPs=")
>>> print(f.decrypt("gAAAAABm1Yv7tyENd3Kbn9E7qPi-YeZUA6SthfP01fUzE0VVT1I_O6z_UUA5Mi0N2Emue6FDboEPEy3AjyvU1iZT8LMCE5UTPg==").decode())
test

 

3. Variables 설정 방법

[Admin] → [Variables]에서 Variables를 추가할 수 있다.

 

파일 선택을 통해 json 파일을 업로드 해 한번에 여러 개의 Variables를 추가하거나, + 버튼을 클릭해 하나씩 추가할 수 있다.

또한 key에 아래의 마스킹 문자열이 포함되면 해당 key의 value는 마스킹된다.

• password
• secret
• passwd
• passwd
• api_key
• apikey
• access_token

[ 파일 선택을 통한 Variables 추가 ]

{
"test_password" : "test",
"test_secret" : "test",
"test_passwd" : "test",
"test_authorization" : "test",
"test_api_key" : "test",
"test_apikey" : "test",
"test_access_token" : "test",
"test_var" : "test"
}

 

위와 같이 json 파일을 생성한 후 파일을 업로드하면 한번에 여러 개의 Variables가 생성되며, 마스킹 문자열이 key에 추가되면 value가 마스킹되어 UI에서 값을 확인할 수 없는 것을 확인할 수 있다.

 

[ + 버튼을 통한 Variables 추가 ]

key, value, description을 작성한 후 save하면 간단하게 Variables를 추가할 수 있다.

 

4. Variables 사용 방법

간단하게 `Variable.get("my_key")`를 사용해 Variable의 값을 가져와 사용할 수 있다.

마스킹 문자열이 포함된 key의 value는 log에도 마스킹되어 출력되며, value가 json 형식인 경우 `deserialize_json=True` 옵션을 사용해 json 형식으로 파싱할 수 있다.

 

• Variable 예제

import pendulum
from airflow.decorators import dag, task
from airflow.models import Variable

@dag(
    schedule="@once",
    start_date=pendulum.datetime(2024, 8, 10, 3, tz=pendulum.timezone("Asia/Seoul")),
    catchup=False,
    tags=["example", "Variables"],
)
def get_Variables():
    
    @task
    def get_Variable():
        test_secret = Variable.get("test_secret")
        test_json_str = Variable.get("test_json")
        test_json = Variable.get("test_json", deserialize_json=True)

        print(f"{type(test_secret)}, {test_secret}")
        print(f"{type(test_json_str)}, {test_json_str}")
        print(f"{type(test_json)}, {test_json}")


    get_Variable()

get_Variables()

 

• 로그

[2024-09-02, 10:39:46 UTC] {logging_mixin.py:190} INFO - <class 'str'>, ***
[2024-09-02, 10:39:46 UTC] {logging_mixin.py:190} INFO - <class 'str'>, {
"a_key" : "a_value",
"b_key" : "b_value"
}
[2024-09-02, 10:39:46 UTC] {logging_mixin.py:190} INFO - <class 'dict'>, {'a_key': 'a_value', 'b_key': 'b_value'}