[Airflow] Airflow에서 DAG 파일을 처리하는 법 (DagFileProcessor)

[Airflow] Airflow에서 DAG 파일을 처리하는 법 (DagFileProcessor)

Airflow는 DAG(Directed Acyclic Graph) 파일을 통해 워크플로우를 정의하는 워크플로우 관리 도구다.

DAG 파일은 Python 코드로 작성되는데, Airflow가 이를 어떻게 인식하고 실행하는 지에 대해 궁금점이 생겼다,

 

Airflow에서 DAG 파일을 처리하는 핵심 컴포넌트 중 하나가 바로 DagFileProcessor다. 이 글에서는 DagFileProcessor가 무엇인지, 어떻게 동작하는지에 대해 알아보고자 한다.

 

DagFileProcessor란?

DagFileProcessor는 Airflow가 DAG 파일을 주기적으로 읽고 파싱하여, 메타 데이터를 저장하게하는 역할을 한다.

Airflow는 단순히 DAG 파일을 실행하는 것이 아니라, DAG 파일에서 유효한 DAG 객체들을 추출하고 이를 Airflow 데이터베이스에 저장해야한다.

 

이를 위해 DagFileProcessor는 주어진 디렉토리에서 DAG 파일을 주기적으로 스캔하고 파싱하여, 데이터베이스에 등록하는 작업을 수행한다. DAG 파일이 변경될 경우에도 이를 감지하여 업데이트 한다.

 

DagFileProcessor는 DAG 파일을 읽고 처리하는 개별적인 작업 단위이기 때문에, Airflow 2.x 부터는 DAG 파싱 성능을 개선하기 위해 멀티프로세스 방식을 도입했으며, DagFileProcessorManager가 여러 개의 프로세서를 관리한다.

 

DagFileProcessor의 동작 방식

Airflow 2.x에서는 다음과 같은 구조로 DAG 파일이 처리된다.

 

1. DagFileProcessorManager 실행

• Airflow 스케줄러가 실행되면 DagFileProcessorManager가 시작됨
• `scheduler.processor_poll_interval` (기본값 1초)마다 DAG 파일을 체크

2. DAG 파일을 여러 개의 DagFileProcessor로 분배

• `dag_file_processor_threads` 설정값에 따라 DAG 파일을 병렬로 처리
• 기본값은 2, DAG가 많다면 4~8로 설정할 수 있음

3. DAG 파일 파싱 및 메타데이터 저장

• DagFileProcessor가 개별 DAG 파일을 실행하여 파싱
• DAG ID, Task 정의, 스케줄 정보 등을 데이터베이스에 저장

4. DAG 변경 감지 및 업데이트

• DAG 파일이 변경되면 이를 감지하여 다시 처리

 

관련 설정

DagFileProcessorManager와 DagFileProcessor의 성능을 최적화하기 위해선 적절한 설정이 필요하다.

해당 설정은 airflow.cfg 파일을 수정하거나 환경 변수를 사용한다.

[scheduler]
# 병렬 처리 개수 조정
# 해당 값을 증가시키면 여러 개의 DAG 파일을 동시에 처리할 수 있다.
dag_file_processor_threads = 4

# Dag 파일 변경 감지 주기 조정
# 기본값은 30초로 DAG 파일 변경을 얼마나 자주 감지할지 설정한다.
min_file_process_interval = 20

# DAG 스캔 주기 조정
# 기본값은 300초로 DAG 디렉토리를 얼마나 자주 스캔할 지 설정한다.
dag_dir_list_interval = 120

 

요약

✅ DagFileProcessor

: DAG 파일을 개별적으로 실행하여 파싱하고, 메타데이터를 저장하는 역할

 

✅ DagFileProcessorManager

: 여러 개의 DagFiileProcessor를 관리하여 병렬 처리를 수행

DAG 변경 사항을 감지하고, DAG 파싱을 최적화함

 

참고

https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/dagfile-processing.html