Airflow 3.x Dag Parsing 최적화 체크리스트

정보공유
1

요즘 llm으로 코드를 작성하고, 리뷰만 진행하다보니 Dag 파싱 최적화는 뒷전이 되어가는 것 같습니다.

Dag 파싱을 최적화 한다는 것?

저만 그런진 모르겠지만 저는 처음에 Dag 가 파싱되는 것을 최적화 한다는 것이 어떤 것인지 이해가 어려웠습니다. 그래서 거기서 부터 설명을 하면 좋을 것 같아서 끄적여보겠습니다.

Dag 파싱은 Airflow의 dag-processor(2.x는 scheduler 내부에서 진행됩니다.)가 주기적으로 Dag 폴더의 모든 .py 파일을 import 해서 Dag 객체를 만들고, 직렬화해서 메타DB(serialized_dag 테이블)에 저장하는 과정입니다.

사용하시는 세팅값마다 다르겠지만 기본적으로는 아래와 같이 세팅되어 있습니다.

  • min_file_process_interval (기본 30s) 마다 각 파일을 다시 파싱
  • dag_dir_list_interval (기본 5분) 마다 파일 목록 재스캔

즉, Dag 파일은 한 번 실행되는 게 아니라 수십 초마다 반복 import 된다고 보시면 됩니다.

그래서 파싱이 느리면 아래와 같은 문제들이 생깁니다.

  • 트리거 → 실행 지연: Asset 트리거하거나 schedule이 와도, 파서가 한 사이클 돌 때까지 task가 안 뜸
  • scheduler heartbeat 지연: 결국 task slot 회수도 늦어짐
  • CPU/메모리 폭증
  • dag_processing.total_parse_time 메트릭이 interval보다 길면 “파서가 못 따라잡는” 상태

비싼 CPU/메모리 위에서 돌아가는 만큼, 한 번은 정리할 가치가 있습니다.

모범 사례가 뭐지?

먼저 무엇이 진짜 모범 사례인지 그래도 공식 문서에 나와있습니다.

표로 보면,

레이어 핵심 권고
코드 Top-level 에서 DB/API/파일 I/O 금지. Variable.get() top-level 금지 → Jinja {{ var.value.X }}. 무거운 import (pandas, torch) 는 task 함수 내부로
파일/디렉터리 .airflowignore 로 비-Dag 파일 명시 제외. 1 파일 1 Dag 원칙 (FileProcessor 병렬화)
인프라 (3.x) dag-processor standalone 서비스 사용. [dag_processor] 섹션의 parsing_processes, min_file_process_interval, file_parsing_sort_mode 튜닝. 3.x 신규 parsing_pre_import_modules 로 공통 모듈 pre-load
회귀 방지 Airflow 3 마이그레이션 위생은 Ruff AIR3 그룹 + 파싱 안티패턴 (top-level Variable.get 등) 은 별도 커스텀 lint

자주 헷갈리는 3.x 변경 사항 4가지

  1. dag-processor: 2.3 에서 optional 도입 (standalone_dag_processor=True) → 3.0 에서 mandatory. 더 이상 scheduler 내부에서 안 돕니다.
  2. 설정 섹션 이동: 파싱 관련 설정이 [scheduler] → [dag_processor] 로 이동. 옛 섹션에 두면 silent 하게 무시.
  3. .airflowignore 기본 syntax: 2.x regex → 3.x glob. 가장 비싼 silent breaking change (뒤에서 자세히).
  4. dag_dir_list_interval deprecated: 2.x 잔재.

본론으로..

막연히 “느린 것 같다” 가 아니라, 어디서부터 어떻게 봐야 하는지 순서대로 정리해봐야합니다.

메트릭으로 현황을 숫자로 박는다

직감 대신 메트릭부터. Airflow 3.x dag-processor 는 StatsD / OpenTelemetry 로 다음 메트릭을 emit 합니다:

  • dag_processing.total_parse_time: 가장 중요. 한 사이클 전체 시간
  • dag_processing.last_duration.<filename>: 파일별 마지막 파싱 시간 — 무거운 파일 식별용
  • dag_processing.last_runtime.<filename>: 파일별 last run 경과
  • dag_processing.processes: 활성 child process 수
  • dag_processing.processor_timeouts: timeout 카운트 (cycle 평균 왜곡 원인)

메트릭 인프라가 없으면 임시로 dag-processor 로그의 file processing timestamp 를 추출해서 cycle 을 근사할 수도 있습니다 (정확하진 않습니다).

제 상황을 실측해보면…

"timestamp":"2026-05-18T03:47:31.967Z"
"timestamp":"2026-05-18T03:50:03.596Z" (+152s)
"timestamp":"2026-05-18T03:52:32.280Z" (+149s)
"timestamp":"2026-05-18T03:55:07.587Z" (+155s)
...
평균: ~155s

설정값 min_file_process_interval=30s 와 ~5배 격차.

여기서 한 가지 중요한 사실은,
min_file_process_interval 은 “이 파일을 마지막 파싱 후 N초가 지나야 재대상에 들어간다” 는 lower bound 이지 cycle 보장이 아닙니다.
cycle 이 155s 면 모든 파일이 30s 를 훨씬 초과해 매번 재파싱 큐에 즉시 재진입 → 30s 설정은 사실상 무의미한 상태가 됩니다.

인프라 설정을 본다 (대부분 여기가 범인입니다)

코드를 의심하기 전에 인프라 한 줄을 먼저 의심하세요. 가장 큰 ROI 가 여기 있는 경우가 많습니다.

# 3.x 는 [dag_processor] 섹션
airflow config get-value dag_processor parsing_processes
airflow config get-value dag_processor min_file_process_interval
airflow config get-value dag_processor file_parsing_sort_mode
  • parsing_processes: 병렬 파서 수. 기본 1. CPU 코어 / 평균 파일 파싱 시간 / 메모리 마진 보고 조정해야합니다.
  • min_file_process_interval: 기본 30s. cycle 이 이보다 길면 의미 없음.
  • file_parsing_sort_mode: 파싱 우선순위. modified_time를 권장합니다.(최근 변경 파일 먼저)

저도 parsing_processes 를 조금 늘려줬습니다.

10:07:57 → 10:09:05 68s
10:09:05 → 10:10:20 75s
10:10:20 → 10:11:29 69s
...
평균: ~72s (표준편차 ~4s)

훨씬 나아졌네요.

공통 import 비용 parsing_pre_import_modules (3.x 신규)

여기서 많이 놓칩니다. 3.x 에 추가된 parsing_pre_import_modules 는 dag-processor manager 가 시작할 때 명시한 모듈을 한 번 import 해두고, child process fork 시 copy-on-write 로 재사용합니다. 모든 Dag가 공통으로 import 하는 패키지가 있다면 직격탄이 됩니다.

# airflow.cfg
[dag_processor]
parsing_pre_import_modules = ...

이런 식으로 모듈을 넣어줄 수 있습니다.
그러면 child 마다 매번 import 하면 N × import 비용이지만, pre-import 하면 1회로 끝.

.airflowignore 와 silent breaking change

코드와 인프라 다음은 파일 디스커버리. 각 번들에 .airflowignore 가 잘 잡혀 있는지:


for d in <여러분들의 번들들>; do
  echo "--- $d ---"
  cat $d/dags/.airflowignore 2>/dev/null || echo "(NONE)"
done

“dag_discovery_safe_mode 만 믿지 말기”

흔히 “dag_discovery_safe_mode=True 면 ‘airflow’/‘DAG’ 키워드 없는 파일은 자동 스킵된다” 고 알려져 있죠.

하지만 utility 파일들이 대부분 from airflow.* import ... 를 포함하므로 키워드 매칭에 걸려 자동 스킵을 우회합니다. 명시 제외가 안전합니다.

.airflowignore 했는데 안 줄어든다?

각 번들에 .airflowignore 를 일괄 작성해서 머지하고 cycle 재측정. 그런데…

12:00:39 → 12:01:46 67s
12:01:46 → 12:03:01 75s
12:03:01 → 12:04:04 63s

헬퍼 파일이 빠졌으면 분명 더 줄어야 하는데.
제 유틸 파일의 dag-processor 로그 timestamp 가 여전히 60 초마다 갱신되고 있었습니다. 파싱이 계속 되고 있다는 뜻이죠.

.airflowignore glob default

직접 Airflow 의 매칭 함수를 호출해서 검증해봤습니다.

from airflow._shared.module_loading import find_path_from_directory
help(find_path_from_directory)
# find_path_from_directory(base_dir_path, ignore_file_name,
#                          ignore_file_syntax: str = 'glob') -> Generator
#                                              ^^^^^^^^^^^^^^^^^^

Airflow 3.x 에서 .airflowignore 기본 syntax 가 glob 으로 변경됐습니다 (2.x 는 regex).
제가 작성한 패턴은 모두 regex 형태 (^_signal_utils\.py$) 라서 glob 으로 해석되면 매칭 0 건. 에러도 안 납니다. 그냥 silent 하게 무시되는 거죠.

실측 비교:

# regexp 모드
list(find_path_from_directory(d, '.airflowignore', ignore_file_syntax='regexp'))

# glob 모드
list(find_path_from_directory(d, '.airflowignore', ignore_file_syntax='glob'))

코드 레벨 안티패턴

여기는 신규 PR 마다 잡으면 누적되지 않는 영역이지만, 한 번은 전수 점검할 가치가 있습니다.

# top-level Variable.get / Connection.get
rg "^(Variable\.get|Connection\.get)" */dags/

# top-level DB 클라이언트 인스턴스화
rg "^(PostgresClient|KISClient)\(" */dags/

# 동적 Dag 생성
rg "for .+ in .+:" */dags/ | rg "DAG\("

이건 CI에서 그래도 칼같이 잡아왔던 터라 모두 양호했습니다.

결론

“Dag 파싱 느려요” 면 보통 코드부터 의심하는데, 이미 코드 위생이 좋은 환경에서는 인프라 탓일 수 있습니다.
측정 없이 직감으로 갔으면 며칠 헛수고할 수도 있습니다 :smiling_face_with_tear:
(하지만 저는 Dag 작성자이자 인프라 관리자이기 때문에 결국 제 탓 ㅠ)

그리고 Airflow 3.x에서의 silent change 들을 알고 있어야할 것 같네요.
다른 분들도 Dag 파싱 최적화 하실 때 체크했던 것들이 있으시면 알려주시면 감사드릴 것 같습니다 :slight_smile: