Skip to content

Airflow DAG 작성

개요

Airflow에서 파이프라인은 DAG(Directed Acyclic Graph) 라는 코드로 표현돼요. 작업 하나가 노드(Node)고, 작업 사이의 의존 관계가 화살표(Edge)죠. "A가 끝나야 B를 돌린다"는 순서가 그래프로 명시되기 때문에, 사람이 순서를 암기하지 않아도 파이프라인 전체가 체계적으로 실행돼요. 공식 문서는 DAG를 "워크플로우를 실행하는 데 필요한 모든 것을 담은 모델"이라고 정의해요. 언제 돌릴지(schedule), 어떤 작업들(tasks), 그 작업들의 순서·조건(dependencies), 완료 시 조치(callbacks)가 이 안에 들어 있어요.

핵심 개념

DAG 선언하기

DAG는 크게 세 가지 방식으로 선언할 수 있어요.

  • with 구문(컨텍스트 매니저) — with DAG(...): 블록 안에 작업을 넣으면 그 블록의 모든 게 자동으로 DAG에 속해요. 가장 많이 쓰는 방식이에요.
  • 표준 생성자 — DAG(...) 객체를 만들고, 각 오퍼레이터에 dag=my_dag로 넘기는 방식.
  • @dag 데코레이터 — 함수를 DAG 생성기로 바꾸는 방식. 함수 밖에서 호출해야 DAG가 로드돼요.

기본 예시는 이렇게 생겼어요.

import datetime
from airflow.sdk import DAG
from airflow.providers.standard.operators.empty import EmptyOperator

with DAG(
    dag_id="my_dag_name",
    start_date=datetime.datetime(2021, 1, 1),
    schedule="@daily",
):
    EmptyOperator(task_id="task")

DAG는 schedule 인자로 실행 주기를, start_date로 시작 시점을 정해요. 실행 주기는 cron 표현(0 0 * * *)이나 프리셋(@daily, @once, @continuous)으로 쓸 수 있어요.

작업 의존성(Dependencies) 선언

의존성 선언의 권장 방식은 >>·<< 연산자예요.

first_task >> [second_task, third_task]
third_task << fourth_task

>>는 앞 작업이 끝나야 뒤 작업이 실행된다는 뜻이에요. 더 명시적인 set_upstream/set_downstream 메서드도 있고, 여러 작업끼리 한꺼번에 묶는 cross_downstream, 이어서 연결하는 chain 헬퍼도 있어요.

DAG 로딩

Airflow는 DAG 번들(Dag bundle) 안의 Python 파일들을 읽어서 실행하고, 그 파일에서 DAG 객체를 로드해요. 이때 파일 최상위(top-level)에 있는 DAG 인스턴스만 발견돼요. 함수 안에서 만든 DAG는 로드되지 않아요. 또 성능 최적화로 airflow와 dag 문자열(대소문자 무시)이 포함된 파일만 검사하는데, 이 동작은 DAG_DISCOVERY_SAFE_MODE 설정으로 끌 수 있어요. .airflowignore 파일로 특정 파일을 로드에서 제외할 수도 있어요.

DAG 실행: Dag Run과 데이터 구간

DAG가 실행될 때마다 Dag Run이라는 인스턴스가 생겨요. 같은 DAG도 여러 Dag Run이 병렬로 돌 수 있고, 각 Dag Run은 data interval(데이터 구간) 을 가져요. 이 구간이 "이 실행이 다루어야 할 데이터의 기간"이에요.

여기서 헷갈리기 쉬운 날짜가 하나 있어요. logical date(논리적 날짜, 예전엔 execution date) 라는 개념인데, 이건 실제로 실행된 시각이 아니라 "이 Dag Run이 스케줄된 의도 시점"이에요. 예를 들어 매일 스케줄이면, logical date는 데이터 구간의 시작을 가리키고 실제 실행 시작은 logical date + 간격이죠. 이 때문에 "왜 오늘 돌렸는데 실행 날짜가 어제지?"라는 혼란이 생겨요.

백필(Backfill)

DAG를 과거 데이터에 대해 다시 돌리고 싶을 때 백필을 써요. 예를 들어 파이프라인을 고쳐서 지난 3개월 데이터를 다시 처리해야 하면, DAG의 과거 Dag Run들을 몽땅 만들어 한 번에 실행할 수 있어요. 각 Dag Run은 그 기간의 날짜 하나를 데이터 구간으로 갖죠.

기본 인자(default_args)와 테스트

여러 작업에 공통으로 쓸 기본값(예: retries)은 DAG에 default_args로 한 번만 넘기면 모든 작업에 자동 적용돼요. 또 dag.test()로 로컬에서 DAG의 실행 흐름을 시뮬레이션할 수 있어서, 스케줄러에 넣기 전에 문법·동작을 검증하기 편해요.

실제 적용 (데이터스케쳐스)

Events 3계층에서 Airflow는 파이프라인 층의 지휘를 맡는 걸로 시나리오를 잡고 있어요. 원천 데이터를 주기적으로 가져와 표준 스키마로 정리하고 창고에 적재하는 흐름을 DAG 하나로 묶는 식이죠. 이 DAG를 작성할 때 가장 먼저 정할 게 작업의 단위(실패가 퍼지지 않게 쪼개는 경계) 와 데이터 구간의 기준이에요.

현재는 역량 단계라서 팀이 다룰 줄 알되 제품 전면 도입 전이에요. DAG를 짤 때는 "어떤 작업을 어떤 순서로"만 신경 쓰면 되고, 실제 실행 환경(스케줄러·워커)과는 분리해서 생각하는 게 좋아요. 실제 실행 단위인 오퍼레이터는 오퍼레이터에서, 언제·얼마나 자주 돌릴지는 스케줄링에서 이어져요.

더 알아보기