Airflow DAG Hash가 계속 바뀌는 문제: Variable.get()의 함정

TTS 배치 DAG의 Task들이 간헐적으로 실패하기 시작했습니다. 30개짜리 Dynamic Task Mapping 중 앞쪽 13개는 성공하고, 뒤쪽 17개가 한꺼번에 실패하는 패턴이었습니다.

DAG 'tts_batch_dag' not found in serialized_dag table

특이한 건 빠르게 끝나는 DAG에서는 재현이 안 되고, 처리량이 많아서 시간이 좀 걸리는 DAG에서만 터진다는 점이었습니다. 처음엔 Worker 리소스 문제인 줄 알았는데, 원인은 전혀 다른 곳에 있었습니다.

에러 상황: Task 실행 중 “DAG not found”

에러 로그를 좀 더 자세히 뜯어보면, Scheduler가 Task를 Worker에 넘기려는 순간 DAG를 찾지 못하고 있었습니다.

scheduler  | ERROR - DAG 'tts_batch_dag' for task instance
  <TaskInstance: ...generate_tts map_index=14 [failed]>
  not found in serialized_dag table

이상한 건 실패한 Task들의 상태였습니다. 성공한 Task에는 hostname=elastic이 찍혀 있는데, 실패한 Task는 전부 hostname=NULL, pid=NULL이었습니다. Worker에 도달조차 못 한 겁니다.

map_index 0-13:  success ✓  (hostname=elastic)
map_index 14+:   failed  ✗  (hostname=NULL, pid=NULL)

DAG 코드에 문법 에러가 있는 것도 아니고, Worker 리소스가 부족한 것도 아니었습니다. 혹시 DAG 자체에 뭔가 이상이 있나 싶어 serialized_dag 테이블을 직접 조회해봤습니다.

SELECT last_updated, dag_hash
FROM serialized_dag
WHERE dag_id = 'tts_batch_dag'
ORDER BY last_updated DESC LIMIT 10;

      last_updated          |             dag_hash
-------------------------------+----------------------------------
 2025-12-09 00:48:24.773546+00 | 5005eb097fc4a58972cf320eac3dc6b7
 2025-12-08 10:27:17.520069+00 | e5db60def7aaa5b89307378afb4ff06c
 2025-12-08 10:14:12.194379+00 | fa0527965698f73c494e901d02ef2ecd
 2025-12-08 10:13:40.324092+00 | b7b40eb57a6e41a0f1160a63551d337a
 2025-12-08 10:13:08.285033+00 | 275b8935d0cb02eebbdbbe8c06bc2903
 2025-12-08 10:12:29.201299+00 | 2718a0e66f733bc924c70a2ea4b95cfa
 2025-12-08 10:11:57.912539+00 | 513c52b38417299fb49a3f48a5c4353e
 2025-12-08 10:11:26.426727+00 | 176e3e5dafbbbbe537340a43f037f416
 2025-12-08 10:10:54.996065+00 | 4f931b60bd4464284471031366dd84d6
 2025-12-08 10:10:22.857835+00 | 0c2ede2f5b25e5a1d974fb7d7fe204e6

DAG 파일을 전혀 수정하지 않았는데, dag_hash가 30초 간격으로 계속 바뀌고 있었습니다. 10개 행이 전부 다른 해시값입니다. Worker가 아니라 DAG 파싱 쪽을 의심할 차례였습니다.

원인 분석: DAG Serialization과 dag_hash의 역할

Airflow의 DAG 처리 흐름

왜 dag_hash가 바뀌면 Task가 실패하는 걸까요? 먼저 Airflow 3.x의 DAG 처리 구조를 짚고 넘어가겠습니다.

Airflow 3.x에서는 scheduler가 DAG 파일을 직접 읽지 않습니다. dag-processor가 DAG 파일을 파싱하고 직렬화(serialize)해서 DB에 저장하면, scheduler가 DB에서 직렬화된 DAG를 읽어 Task를 스케줄링하고 worker가 실행합니다. 3.x부터는 이 dag-processor가 별도 프로세스로 반드시 떠 있어야 합니다. 공식 업그레이드 가이드도 “로컬이나 개발 환경에서도 dag-processor를 따로 띄워야 한다”고 안내합니다.

여기서 핵심은 dag-processor가 DAG 파일을 반복 파싱한다는 점입니다. 주기를 정하는 값은 [dag_processor] min_file_process_interval이고 기본값이 30초입니다(2.x에서는 같은 이름의 옵션이 [scheduler] 섹션에 있었습니다). 파싱할 때마다 직렬화 결과를 해시로 만들어 dag_hash에 저장하는데, DAG 파일이 변하지 않았다면 이 해시값은 항상 같아야 합니다.

dag_hash가 바뀌면 왜 Task가 실패할까?

Airflow 3.0의 DAG 버저닝(AIP-66)에서는 직렬화 결과가 달라질 때마다 새로운 DAG 버전이 만들어지고, DAG Run은 시작 시점의 버전에 묶여 끝까지 실행됩니다. 문제는 DAG Run이 실행되는 도중에 직렬화 결과가 계속 바뀔 때 생깁니다. 실제로 겪었던 시나리오를 시간순으로 재구성하면 이렇습니다.

Airflow 스케줄러는 큐에 넣을 Task를 고를 때 그 DAG Run에 해당하는 직렬화 DAG를 DB에서 다시 조회합니다. 조회에 실패하면 DAG '...' for task instance ... not found in serialized_dag table을 로그에 남기고, 해당 DAG의 SCHEDULED 상태 Task를 한꺼번에 FAILED로 바꿉니다. Worker에 넘기기 전 단계에서 끝나버리므로 hostname과 pid가 NULL로 남습니다. 실행 시간이 짧은 DAG에서는 재파싱이 일어나기 전에 모든 Task가 끝나므로 문제가 드러나지 않고, 오래 걸리는 DAG에서만 터졌던 이유가 이겁니다.

근본 원인: DAG 파일 최상위의 Variable.get()

그러면 DAG 파일을 수정하지 않았는데 왜 dag_hash가 매번 달라졌을까요? 문제는 DAG 파일의 최상위 스코프에서 Variable.get()을 호출하는 코드에 있었습니다.

# ❌ DAG 파일의 최상위 스코프: DAG parsing 시 매번 실행됨

from airflow.sdk import Variable

# DAG 함수 바깥에서 Variable.get() 호출!
gemini_manager = GeminiManager(
    api_key=Variable.get("GOOGLE_API_KEY"),  # ← 이게 문제
    ftp_manager=ftp_manager,
)

@dag(...)
def tts_batch_pipeline():
    result_batches = generate_tts.partial(
        gemini_manager=gemini_manager,  # 위에서 생성한 Manager를 그대로 전달
    ).expand(...)

Airflow의 dag-processor는 DAG 파일을 파싱할 때 파일의 모든 최상위 코드를 실행합니다. 즉, Variable.get("GOOGLE_API_KEY") 호출이 30초마다 반복되는 셈입니다.

여기서 바로 해시가 달라지는 건 아닙니다. dag_hash는 직렬화된 DAG를 JSON으로 정렬해 만든 md5 값이라서, 직렬화 결과가 같으면 몇 번을 호출하든 해시도 같습니다. 문제는 그 다음 줄입니다. 최상위에서 만든 gemini_manager 객체를 partial()로 Task에 넘기고 있었고, 이 객체가 직렬화 대상에 포함됐습니다.

Airflow의 직렬화기는 자기가 아는 타입이 아니면 마지막에 str()로 떨어뜨립니다. __repr__을 따로 정의하지 않은 객체라면 그 결과는 <myapp.GeminiManager object at 0x7f3d2c1a4e50> 같은 문자열이고, 여기에는 파싱마다 달라지는 메모리 주소가 들어 있습니다. 직렬화 JSON이 달라지니 md5도 달라지고, 결국 dag_hash가 30초마다 새 값이 됩니다.

해결: Variable.get()을 Task 실행 시점으로 이동

해결 방법은 명확합니다. DAG 파일 최상위에서 Variable.get()을 호출하는 코드를 Task 함수 내부로 옮기면 됩니다.

# ✅ After: DAG 파일에서는 Manager를 생성하지 않음

@dag(...)
def tts_batch_pipeline():
    result_batches = generate_tts.partial(
        requests_url=...,   # 정적 데이터만 전달
    ).expand(...)

# ✅ Task 파일: 실행 시점에만 Variable 접근

@task
def generate_tts(requests_url, batch_idx, batch_size):
    # Task 실행 시에만 Variable 접근!
    from airflow.sdk import Variable

    api_key = Variable.get("GOOGLE_API_KEY")
    gemini_manager = GeminiManager(api_key=api_key, ...)

    # 작업 수행
    return run_generate_tts(...)

핵심은 Variable.get()을 @task 데코레이터가 붙은 함수 안으로 옮긴 것입니다. 이렇게 하면 dag-processor가 DAG를 파싱할 때는 Variable에 접근하지 않으므로, 직렬화 결과가 항상 동일해집니다.

수정 배포 후 같은 쿼리를 다시 돌려봤습니다. 차이가 분명합니다.

-- Before: dag_hash가 매번 다름
10:10:22 - 0c2ede2f5b25e5a1d974fb7d7fe204e6
10:10:54 - 4f931b60bd4464284471031366dd84d6
10:11:26 - 176e3e5dafbbbbe537340a43f037f416

-- After: dag_hash가 고정!
10:27:17 - e5db60def7aaa5b89307378afb4ff06c
10:28:00 - e5db60def7aaa5b89307378afb4ff06c ✓
10:29:00 - e5db60def7aaa5b89307378afb4ff06c ✓

dag_hash가 고정되면 dag-processor는 “직렬화 결과가 그대로다”라고 판단하고 DB 쓰기 자체를 건너뜁니다. 새 DAG 버전이 만들어지지 않으니 실행 중인 DAG Run이 흔들릴 일도 없습니다.

DAG Parsing vs Task Execution: 반드시 알아야 할 규칙

이 문제의 본질은 “DAG 파싱 시점에 실행되는 코드”와 “Task 실행 시점에 실행되는 코드”의 구분입니다. Airflow를 처음 쓸 때 가장 헷갈리는 부분이기도 합니다.

parsing 시점에 Variable에 접근하는 코드는 2.x에서도 3.x에서도 그대로 동작합니다. 달라진 것은 대가입니다. 3.x는 직렬화 결과가 바뀔 때마다 DAG 버전을 새로 만들고 실행 중인 DAG Run을 특정 버전에 묶기 때문에, 파싱마다 흔들리는 값 하나가 진행 중인 Run 전체를 무너뜨릴 수 있습니다.

파싱 단계에서 Variable 조회 자체를 줄여야 한다면 [secrets] use_cache를 켜는 선택지도 있습니다. DAG 파싱 구간에 한해 Variable을 로컬 캐시하는 실험적 옵션이라, 값 변경이 반영되기까지 cache_ttl_seconds만큼 지연됩니다. 다만 이 옵션은 조회 횟수를 줄여줄 뿐 해시가 흔들리는 원인을 없애주지는 않습니다.

피해야 할 패턴

DAG 파일의 최상위 스코프(즉 @task 바깥)에서 피해야 할 것들을 정리하면 이렇습니다.

# ❌ DAG parsing 시 실행되는 최상위 코드들: 전부 문제!

api_key = Variable.get("API_KEY")             # 파싱마다 DB 조회
conn = BaseHook.get_connection("my_conn")     # 파싱마다 DB 조회
start_date = datetime.now()                   # 매번 다른 값 → 해시가 바뀜
data = requests.get("https://api.com/...")    # 파싱마다 네트워크 호출
config = json.load(open("/tmp/config.json"))  # 파일이 바뀌면 해시가 바뀜

위쪽 두 줄은 파싱마다 DB를 두드리는 비용 문제이고, 아래 세 줄은 그 값이 DAG 정의에 흘러 들어가면 곧바로 해시를 흔듭니다. 전부 @task 함수 안으로 옮기는 편이 안전합니다. DAG 파일의 최상위에는 정적 정의만 두세요.

# ✅ DAG 파일: 정적 정의만

@dag(
    schedule="@daily",
    start_date=datetime(2025, 1, 1),  # 고정값은 OK
    catchup=False,
)
def my_pipeline():
    result = my_task.partial(
        static_param="value",  # 정적 값만 전달
    ).expand(...)

# ✅ Task 파일: 동적 리소스는 여기서

@task
def my_task(static_param):
    from airflow.sdk import Variable

    api_key = Variable.get("API_KEY")
    conn = BaseHook.get_connection("my_conn")
    manager = MyManager(api_key, conn)

    return manager.process(static_param)

마치며

디버깅 과정에서 가장 도움이 됐던 건 serialized_dag 테이블을 직접 조회해본 것이었습니다. dag_hash가 고정되어 있는지 확인하는 것만으로도 이 유형의 문제를 빠르게 진단할 수 있으니, Airflow에서 원인 모를 Task 실패가 발생하면 가장 먼저 확인해볼 만한 지점입니다.

참고자료

• Airflow 공식 문서: Best Practices (Top level Python Code)
• Airflow 공식 문서: DAG Serialization
• Airflow 공식 문서: Configuration Reference ([dag_processor] min_file_process_interval)
• Airflow 공식 문서: Dag Bundles
• Airflow 공식 문서: Upgrading to Airflow 3
• Apache Airflow 3 is Generally Available (DAG versioning)