개요
Airflow 스케쥴러의 DagFileProcessor는 DAG 정의 파일을 처리해 JSON 형식으로 직렬화한 후, Airflow MetaData DB SerializedDagModel에 저장한다.
이 작업은 일정 주기마다 반복적으로 수행되는 것으로 보이는데, 최근에 관련 설정을 확인하게 되어서 기록할 겸 DAG 파일 처리에 대해서 정리한다.
DAG 파일 처리
DAG 파일 처리는 dag_folder에 저장된 Dag 정의 파일을 DAG 객체로 변환하는 작업으로, 아래의 두 개 프로세스가 동작하여 수행한다.
- DagFileProcessorManager : 무한 루프로 실행되는 프로세스로, 처리해야 할 파일을 결정한다.
- DagFileProcessorProcess : 개별 파일을 DAG 객체로 변환한다.
DagFileProcessorManager는 사용자 코드를 실행할 수 있다. 따라서 스케쥴러 프로세스와 같은 호스트가 아니라 다른 호스트에서 실행하도록 설정할 수도 있는데, 이런 경우 AIRFLOW__SCHEDULER__STANDALONE_DAG_PROCESSOR 설정을 True로 변경해야 한다. 이후 airflow dag-processor 명령어로 DagFileProcessorManager를 실행할 수 있다.
그 외의 경우에는 스케쥴러를 시작해야 DagFileProcessorManager도 함께 실행된다.
동작 흐름
DagFileProcessorManager 동작 순서
1. 새 파일 확인 : DAG를 마지막으로 새로고침하고 경과된 시간이 dag_dir_list_interval 보다 큰 경우, dag_folder 내의 파일 목록을 업데이트한다.
2. 최근 처리한 파일 제외 : min_file_process_interval 보다 최근에 처리되었고 수정되지 않은 파일을 목록에서 제외한다.
3. file path 큐에 추가 : 확인한 파일을 큐에 추가한다.
4. 파일 처리 : 각 파일을 대상으로 DagFileProcessorProcess를 시작한다.
5. 결과 수집 : 완료된 DAG processor로부터 결과를 수집한다.
6. 로그 통계 : 통계를 출력하고 dag_processing.total_parse_time 값을 반환한다.
일련의 과정은 무한히 반복된다.
DagFileProcessorProcess 동작 순서
1. 파일 처리 : dag_file_processor_timeout 내에 전체 처리를 완료한다.
2. 모듈 로드 : DAG 파일을 python 모듈로 로드한다. dagbag_import_timeout 내에 완료되어야 한다.
3. 모듈 처리 : python 모듈 내 DAG 객체를 찾는다.
4. DagBag 반환 : DAG 객체 목록을 DagFileProcessorManager에 제공한다.
관련 설정
[core]
# DagFileProcessor의 타임아웃 시간
dag_file_processor_timeout = 50
# python file 임포트 타임아웃 시간
dagbag_import_timeout = 30.0
[scheduler]
# DAG 디렉터리 스캔 빈도
dag_dir_list_interval = 300
# DAG 파일 처리 빈도
min_file_process_interval = 30min_file_process_interval의 경우 값이 낮으면 CPU 사용량이 증가하니 서버에 부하를 주지 않을 정도로 적절하게 조정해야 한다.
참고 문서
https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#dag-dir-list-interval