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

DAG 'tts_batch_dag' not found in serialized_dag table

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

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

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

09:45:24  DAG Run 시작 (dag_hash: c60abd28)
          ├── Task 0~13 실행 시작 → 정상 완료 ✓

09:45:55  dag-processor가 DAG 재파싱
          └── 새로운 dag_hash 생성 (00f68c14) → DB 덮어쓰기

09:46:28  scheduler가 Task 14~29 실행 시도
          ├── Task들은 이전 dag_hash (c60abd28)를 참조
          ├── DB에는 새 dag_hash (00f68c14)만 존재
          └── ❌ "DAG not found" → Task 전부 실패!

# ❌ 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(
        tts_manager=tts_manager,  # 위에서 생성한 Manager 전달
    ).expand(...)

# ✅ 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(...)

-- 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 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/...")  # 외부 API → 응답이 직렬화에 포함될 수 있음
config = json.load(open("/tmp/config.json"))  # 파일 내용 변경 시 hash 변경

# ✅ 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)