Airflow

스케줄러가 top-level 코드를 반복 실행하는 이유

Operator를 만들고 Dag 관계를 구성하는 데 필요하지 않은 top-level 코드는 최대한 작성하지 않아야 한다. 이는 Airflow 스케줄러의 설계 방식과, top-level 코드 파싱 속도가 Airflow의 성능·확장성에 미치는 영향 때문이다.

Airflow 스케줄러는 Operator의 ``execute`` 메서드 바깥 코드를 최소 :ref:`min_file_process_interval<config:dag_processor__min_file_process_interval>` 초 간격으로 실행한다. 이는 스케줄링과 의존성이 시간에 따라 바뀔 수 있는 동적 스케줄링을 지원하기 위함이며, 스케줄러는 Dag 파일 내용이 스케줄된 태스크에 계속 올바르게 반영되도록 지속적으로 확인한다.

구체적으로 DB 접근, 무거운 연산, 네트워킹 작업은 top-level에서 실행하면 안 된다. 특히 놓치기 쉬운 요인 중 하나가 top-level import다 - 무거운 import는 예상보다 훨씬 큰 오버헤드를 만들 수 있으며, 이를 피하려면 Python 콜러블 내부의 local import로 옮기면 된다. 실제 예시에서 ``expensive_api_call``을 top-level에서 호출하면 그 함수 하나 때문에 함수적으로 동일한 Dag 대비 파싱에 1000초가 추가로 걸릴 수 있다 - 그 호출을 태스크의 context 안(``execute``가 실제로 실행되는 시점)으로 옮기면 이 오버헤드가 사라진다.

import 문 역시 top-level 코드로 집계된다는 점에 유의해야 한다. import 자체가 오래 걸리거나, import된 모듈이 자체적으로 top-level 코드를 실행하는 경우 스케줄러 성능에 영향을 줄 수 있다. ``pandas``, ``torch``, ``tensorflow`` 같은 무거운 라이브러리는 top-level import 대신, 그것을 실제로 사용하는 태스크 함수 내부에서 import하는 방식으로 우회할 수 있다.

내 코드가 top-level인지 확인하는 방법

어떤 코드가 top-level인지 판별하려면 Python 파싱의 세부 동작을 이해해야 한다. 일반적으로 Python은 파일을 파싱할 때 보는 코드를 실행하며, 메서드 내부 코드는 (일반적으로) 실행하지 않는다. 다만 메서드의 기본값(default value)을 결정하는 코드처럼, 직관적이지 않은 특수 케이스들도 top-level 코드로 취급된다.

실용적으로 판별하는 방법은 코드에 print 문을 추가하고 ``python <my_dag_file>.py``로 직접 실행해 어떤 print가 출력되는지 보는 것이다. 문서의 예시에서 ``get_task_id()``는 ``PythonOperator(task_id=get_task_id(), ...)``처럼 인자 값을 만드는 데 쓰이므로 파싱 시점에 실행되어 "Executing 1"이 출력되지만, ``python_callable=get_array``로 전달된 ``get_array``는 함수 자체가 아니라 참조만 전달되므로 파싱 시점에는 호출되지 않아 "Executing 2"는 출력되지 않는다.

코드 품질 관리 - Ruff AIR 규칙, mypy 플러그인

Ruff는 이 문서의 모범 사례 중 일부를 ``AIR`` 규칙으로 탐지·강제해준다. 여기에는 지원 종료된(deprecated) 패턴과 마이그레이션 이슈도 포함된다. ``ruff>=0.16.4``를 설치한 뒤 ``ruff check dags/ --select AIR3`` 명령으로 ``dags/`` 디렉터리의 Dag들을 분석해 관련 규칙 위반을 리포트할 수 있다.

예를 들어 Airflow 2 스타일의 legacy Dag(``@dag()`` 데코레이터에 schedule 인자 없음, ``airflow.datasets.Dataset`` import, ``airflow.sensors.filesystem.FileSensor`` import)를 검사하면 다음과 같은 진단이 나온다: ``AIR301``(schedule 인자 명시 필요), ``AIR302``(schedule_interval/Dataset이 Airflow 3.0에서 제거됨), ``AIR303``(FileSensor가 Airflow 3.0에서 standard 프로바이더로 이동함). Ruff를 개발 워크플로에 통합하면 deprecation을 사전에 능동적으로 해결하고 버전 간 전환을 매끄럽게 할 수 있다.

새로운 Airflow 모범 사례를 제안하고 이에 대응하는 Ruff AIR 규칙을 추가하고 싶다면, 정해진 기여자 프로세스(``24_proposing_best_practices_and_air_rules.rst``)를 따르면 된다.

``mypy``로 Dag를 타입 체크하는 경우, 선택적 패키지인 ``apache-airflow-mypy`` 플러그인을 사용하면 타입 데코레이터나 오퍼레이터 출력 같은 Airflow 특유의 패턴에 대해 정확한 타입 체크 결과를 얻을 수 있다.