기본 구조
Airflow의 구성 요소를 설명해보세요
- Scheduler — DAG를 파싱하고 실행할 태스크를 판단해 큐에 넣습니다
- Executor — 태스크를 실제로 실행할 방식을 결정합니다
- Worker — 태스크를 수행합니다(실행자에 따라 존재 여부가 다름)
- Metadata DB — DAG·태스크 상태, 실행 이력, 연결 정보를 저장합니다
- Webserver — UI
핵심은 메타데이터 DB가 단일 진실 공급원이라는 점입니다. 스케줄러도 워커도 여기를 보고 판단하므로, DB 성능이 전체 처리량을 좌우합니다.
DAG는 무엇이고 왜 순환이 없어야 하나요
방향성 비순환 그래프(Directed Acyclic Graph)입니다. 태스크가 노드, 의존 관계가 엣지입니다.
순환이 있으면 어느 태스크부터 실행할지 결정할 수 없습니다. A가 B를 기다리고 B가 A를 기다리면 영원히 시작하지 못합니다. Airflow는 DAG 파싱 시 순환을 감지해 오류를 냅니다.
DAG 파일이 파싱되는 방식을 설명해보세요
스케줄러가 DAG 폴더의 파이썬 파일을 주기적으로 실행해 DAG 객체를 수집합니다. 이게 몇 가지 함정을 만듭니다.
최상위 코드는 파싱할 때마다 실행됩니다. DAG 파일 상단에서 DB를 조회하거나 API를 호출하면 수십 초마다 그 호출이 일어납니다.
# 나쁨 — 파싱마다 실행됨
rows = db.query("SELECT ...")
with DAG(...) as dag:
for r in rows:
PythonOperator(task_id=f"t_{r.id}", ...)
# 좋음 — 실행 시점에만 동작
def fetch_and_process(**context):
rows = db.query("SELECT ...")
...
with DAG(...) as dag:
PythonOperator(task_id="process", python_callable=fetch_and_process)
파싱 시간이 길어지면 스케줄러 전체가 느려집니다. dagbag_import_timeout을 넘기면 아예 DAG가 사라집니다.
스케줄링
execution_date(logical date)는 무엇인가요
Airflow에서 가장 헷갈리는 개념입니다. 처리 대상 기간의 시작점을 가리키며, 실제 실행 시각이 아닙니다.
schedule="@daily"인 DAG에서 2026-08-11 실행은 2026-08-12 00:00에 시작됩니다. 8월 11일 하루치 데이터가 다 모인 뒤에 처리하는 것이 자연스럽기 때문입니다.
데이터 구간: [2026-08-11 00:00, 2026-08-12 00:00)
logical date: 2026-08-11
실제 실행: 2026-08-12 00:00
이 설계 덕분에 백필과 정기 실행이 같은 코드로 동작합니다. 날짜를 파라미터로 받으니 과거 어느 날짜든 똑같이 돌릴 수 있습니다.
Airflow 2.2부터는 data_interval_start / data_interval_end라는 더 명확한 이름이 생겼습니다.
코드에 CURRENT_DATE를 쓰면 왜 안 되나요
백필이 불가능해지기 때문입니다. 과거 날짜로 실행해도 쿼리는 항상 오늘을 보게 됩니다.
-- 나쁨 — 언제 돌려도 오늘 기준
WHERE dt = CURRENT_DATE - 1
-- 좋음 — 실행 컨텍스트의 날짜를 주입
WHERE dt = '{{ ds }}'
주요 템플릿 변수입니다.
| 변수 | 의미 |
|---|---|
{{ ds }} |
logical date (YYYY-MM-DD) |
{{ ds_nodash }} |
YYYYMMDD |
{{ data_interval_start }} |
데이터 구간 시작 |
{{ data_interval_end }} |
데이터 구간 끝 |
{{ prev_ds }} |
이전 실행의 ds |
catchup은 무엇인가요
start_date부터 현재까지 밀린 실행을 자동으로 채우는 기능입니다. 기본값이 True라, 과거 날짜로 start_date를 잡고 DAG를 켜면 수백 개 실행이 한꺼번에 시작됩니다.
with DAG(
dag_id="daily_agg",
start_date=datetime(2024, 1, 1),
schedule="@daily",
catchup=False, # 밀린 것 무시, 다음 주기부터
max_active_runs=1, # 백필할 때도 동시 실행 제한
):
과거 데이터를 채워야 한다면 catchup=True로 두되 max_active_runs로 동시성을 제한해야 소스 시스템이 버팁니다.
DAG 간 의존성은 어떻게 표현하나요
- TriggerDagRunOperator — 다른 DAG를 직접 트리거. 밀어내는(push) 방식
- ExternalTaskSensor — 다른 DAG의 태스크 완료를 기다림. 당기는(pull) 방식. 스케줄 주기가 다르면 시각 매칭이 까다롭습니다
- Dataset(데이터 인식 스케줄링) — 업스트림이 데이터셋을 갱신하면 다운스트림이 자동 실행. Airflow 2.4+
Dataset 방식이 가장 깔끔합니다. 시각이 아니라 데이터 준비 여부로 트리거되므로 스케줄 간 불일치 문제가 없습니다.
orders = Dataset("s3://warehouse/orders")
# 생산자
with DAG("load_orders", ...):
PythonOperator(task_id="load", outlets=[orders], ...)
# 소비자 — orders 가 갱신되면 실행
with DAG("build_mart", schedule=[orders], ...):
...
태스크
XCom은 무엇이고 무엇을 주의해야 하나요
태스크 간에 작은 값을 주고받는 메커니즘입니다. 메타데이터 DB에 저장되므로 큰 데이터를 넣으면 안 됩니다.
파일 경로나 식별자만 넘기고 실제 데이터는 오브젝트 스토리지에 두는 것이 원칙입니다.
# 나쁨 — DataFrame 을 XCom 에
return df
# 좋음 — 경로만 넘김
df.to_parquet(path)
return path
Operator, Sensor, Hook을 구분해보세요
- Operator — 하나의 작업 단위.
PythonOperator,BashOperator,SparkSubmitOperator - Sensor — 조건이 충족될 때까지 기다리는 특수한 Operator. 파일 도착, 파티션 생성, 외부 태스크 완료
- Hook — 외부 시스템 접속을 감싼 인터페이스. Operator 내부에서 씁니다.
PostgresHook,S3Hook
Sensor가 워커를 잡아먹는 문제는 어떻게 해결하나요
기본 모드(poke)의 Sensor는 대기하는 동안 워커 슬롯을 계속 점유합니다. 센서가 많으면 실제 작업이 실행되지 못하고 데드락처럼 보이는 상황이 생깁니다.
mode="reschedule"로 바꾸면 poke 사이에 슬롯을 반납합니다. 대기 시간이 길면 반드시 이 모드를 써야 합니다.
S3KeySensor(
task_id="wait_file",
bucket_key="s3://bucket/data/{{ ds }}/_SUCCESS",
mode="reschedule", # 대기 중 슬롯 반납
poke_interval=300,
timeout=60 * 60 * 6, # 타임아웃 필수
)
Airflow 2.2+의 Deferrable Operator는 더 나아가, 대기를 Triggerer 프로세스에 넘기고 워커를 완전히 비웁니다. 대기가 많은 파이프라인에서 자원 효율이 크게 좋아집니다.
timeout을 안 걸면 조건이 영영 충족되지 않을 때 무한정 대기합니다.
실행자(Executor) 선택 기준은 무엇인가요
| Executor | 특징 |
|---|---|
| Sequential | 하나씩. 개발용 |
| Local | 단일 머신 멀티프로세스. 소규모 |
| Celery | 워커 풀 + 메시지 브로커. 시작 빠름, 환경 고정 |
| Kubernetes | 태스크마다 파드. 격리·유연하지만 시작 지연 |
| CeleryKubernetes | 둘 혼용 |
짧은 태스크가 많으면 Celery, 태스크마다 다른 의존성·리소스가 필요하면 Kubernetes가 맞습니다. Kubernetes 실행자는 파드 시작에 수 초~수십 초가 걸려서, 1초짜리 태스크 1000개에는 부적합합니다.
운영
태스크를 멱등하게 만들어야 하는 이유는 무엇인가요
재시도, 백필, 수동 재실행이 일상적으로 일어나기 때문입니다. 같은 logical date로 여러 번 돌려도 결과가 같아야 안심하고 재실행할 수 있습니다.
# 나쁨 — 재실행하면 중복 적재
INSERT INTO target SELECT * FROM source WHERE dt = '{{ ds }}'
# 좋음 — 해당 파티션을 지우고 다시 씀
DELETE FROM target WHERE dt = '{{ ds }}';
INSERT INTO target SELECT * FROM source WHERE dt = '{{ ds }}';
재시도는 어떻게 설정하나요
default_args = {
"retries": 3,
"retry_delay": timedelta(minutes=5),
"retry_exponential_backoff": True,
"max_retry_delay": timedelta(hours=1),
"execution_timeout": timedelta(hours=2),
}
execution_timeout이 중요합니다. 없으면 멈춘 태스크가 영원히 running 상태로 남아 슬롯을 잡습니다.
그리고 재시도가 의미 있는 실패인지 구분해야 합니다. 일시적 네트워크 오류는 재시도가 맞지만, 잘못된 스키마나 권한 문제는 몇 번을 돌려도 실패합니다. 이런 경우 재시도는 알림만 늦출 뿐입니다.
동시성을 제어하는 설정에는 무엇이 있나요
| 설정 | 범위 |
|---|---|
parallelism |
전체 클러스터의 동시 태스크 수 |
max_active_runs |
DAG 하나의 동시 실행 수 |
max_active_tasks (구 concurrency) |
DAG 하나의 동시 태스크 수 |
max_active_tis_per_dag |
특정 태스크의 동시 실행 수 |
| Pool | 자원별 슬롯 제한 |
Pool이 실무에서 특히 유용합니다. DB 커넥션이 10개뿐이라면 그 DB를 쓰는 태스크들을 슬롯 10짜리 풀에 넣어 과부하를 막습니다.
PythonOperator(task_id="query_db", pool="warehouse_pool", ...)
스케줄러가 느려지는 원인은 무엇인가요
- DAG 파싱 시간 — 최상위 코드에서 무거운 작업을 하는 경우
- DAG 파일 수 — 파일이 수천 개면 파싱 주기가 길어집니다
- 메타데이터 DB 부하 — 태스크 인스턴스 기록이 수백만 건 쌓이면 쿼리가 느려집니다
- 동적 DAG 생성 남용 — 실행 때마다 구조가 바뀌면 부담이 큽니다
메타데이터는 airflow db clean으로 주기적으로 정리해야 합니다. 로그도 함께 쌓이므로 보관 정책이 필요합니다.
태스크가 계속 queued 상태입니다. 원인이 무엇일까요
스케줄러는 실행을 결정했는데 워커가 못 집어가는 상태입니다.
- 워커 부족 — 워커가 죽었거나 개수가 모자람
- Pool 슬롯 소진 — 해당 풀이 꽉 참
- 동시성 상한 도달 —
parallelism,max_active_tasks - 큐 불일치 — 태스크가 지정한 큐를 처리하는 워커가 없음
- K8s 실행자에서 파드 Pending — 노드 자원 부족
UI의 Pool 화면과 워커 로그를 함께 봐야 원인이 좁혀집니다.
DAG를 어떻게 테스트하나요
# 파싱 오류 확인 — 가장 먼저
python dags/my_dag.py
# DAG 구조 확인
airflow dags list
airflow tasks list my_dag --tree
# 태스크 하나만 실행 (의존성 무시)
airflow tasks test my_dag my_task 2026-08-11
# DAG 전체를 특정 날짜로 실행
airflow dags test my_dag 2026-08-11
CI에서는 최소한 모든 DAG 파일의 임포트 검증과 순환 검사를 돌리는 것이 좋습니다. 배포 후에 파싱 오류를 발견하면 그 DAG가 UI에서 사라져버립니다.
Airflow가 적합하지 않은 경우는 언제인가요
- 초 단위 지연이 필요한 스트리밍 — Airflow는 배치 오케스트레이터입니다
- 아주 잦은 실행(분 단위 이하) — 스케줄러 오버헤드가 큽니다
- 단순 크론 하나 — 관리 부담만 늘어납니다. Cloud Scheduler 같은 것으로 충분합니다
- 데이터 처리 자체 — Airflow는 조율하는 도구입니다. 워커에서 대용량을 직접 처리하지 말고 Spark·BigQuery에 넘겨야 합니다