DAG 실행

DAG 실행 (Dag Runs)

Airflow의 DAG는 정의만으로는 아무 일이 일어나지 않아요. DAG가 실제로 실행되는 순간, 그 시간대의 인스턴스가 하나 생기는데 이걸 DAG 실행(Dag Run)이라고 불러요. 실행할 때마다 Dag Run이 새로 만들어진다 보면 돼요. 여러 개의 Dag Run은 서로 독립적으로 돌아가서, 같은 DAG를 동시에 여러 번 실행해도 서로 간섭하지 않아요.

출처: Dag Runs - Apache Airflow 공식 문서

Dag Run 상태 (Dag Run Status)

DAG가 끝까지 실행되면 그 결과로 Dag Run에 상태가 정해져요. DAG 안의 태스크들과 그 의존 관계가 실행 결과를 좌우하죠. 모든 태스크가 더 이상 다른 상태로 전이할 수 없는 종료 상태, 예를 들어 success, failed, skipped에 도달했을 때 Dag Run 상태가 확정돼요. 특히 상태 판정은 '리프 노드(leaf nodes)', 즉 자식 태스크가 없는 마지막 태스크들의 상태를 기준으로 이뤄져요.

Dag Run이 가질 수 있는 종료 상태는 두 가지예요.

• 모든 리프 노드 상태가 success 또는 skipped면 Dag Run은 success
• 리프 노드 중 하나라도 failed 또는 upstream_failed면 Dag Run은 failed

한 가지 조심할 점이 있어요. 일부 태스크에 트리거 규칙(trigger rule)을 지정해 두면 예상 밖의 동작이 나올 수 있어요. 예를 들어 리프 태스크의 트리거 규칙이 all_done이라면 다른 태스크 상태와 무관하게 그 태스크는 무조건 실행돼요. 그 태스크가 성공하면, 중간에 뭔가 실패했더라도 전체 Dag Run이 success로 표시될 수 있으니 주의해야 해요.

Airflow 2.7에서 추가 — 현재 실행 중인 Dag Run이 있는 DAG는 UI 대시보드의 "Running" 탭에서 볼 수 있고, 최근 Dag Run이 실패로 끝난 DAG는 "Failed" 탭에서 찾을 수 있어요.

데이터 구간 (Data Interval)

Airflow의 모든 Dag Run에는 '데이터 구간(data interval)'이 할당돼요. 이건 해당 실행이 다루는 시간 범위를 뜻하죠. 예를 들어 @daily로 스케줄된 DAG라면 데이터 구간이 매일 자정(00:00)에 시작해 다음 자정(24:00)에 끝나요.

보통 Dag Run은 데이터 구간이 끝난 뒤에 스케줄돼요. 그래야 그 시간 동안 모인 데이터를 모두 취합할 수 있으니까요. 즉 2020-01-01 하루치 데이터를 다루는 실행은 2020-01-01이 지난 뒤, 다시 말해 2020-01-02 00:00:00 이후에야 시작돼요.

Airflow의 모든 날짜는 어떤 식으로든 이 데이터 구간 개념과 연결돼 있어요. Dag Run의 '논리적 날짜(logical date)'는 실제 실행 시점이 아니라 데이터 구간의 시작을 가리켜요. Airflow 2.2 이전 버전에서는 이걸 execution_date라고 불렀어요.

마찬가지로 DAG와 태스크의 start_date 인자도 같은 논리적 날짜를 가리켜요. 즉 이 값은 'DAG의 첫 번째 데이터 구간 시작'이지, 태스크가 실제로 실행되기 시작하는 시점이 아니에요. 그래서 Dag Run은 start_date로부터 한 구간이 지난 뒤에야 처음 스케줄되는 거예요.

수동 트리거와 데이터 구간

UI, CLI, REST API, TriggerDagRunOperator 등으로 DAG를 수동 트리거할 때, 실행의 data_interval이 입력한 logical_date로부터 자동으로 유도되거나 같다고 가정하면 안 돼요.

스케줄 실행에서는 타임테이블이 데이터 구간을 직접 정의해요. 하지만 수동 트리거 실행에서는 결과 data_interval이 타임테이블과 트리거 경로에 따라 달라져서, 실행의 logical_date와 다를 수 있어요.

DAG 로직에서 수동 실행에 대해 사용자가 지정한 날짜가 필요하다면, data_interval_start나 data_interval_end와 같다고 가정하지 말고 logical_date를 명시적으로 사용하는 게 안전해요. 업그레이드 지침은 Manual Dag Runs and logical_date를 참고하세요.

DAG 다시 실행하기

DAG를 다시 실행하고 싶은 경우가 생겨요. 대표적인 게 스케줄 실행이 실패했을 때죠.

Catchup

start_date(그리고 선택적으로 end_date)를 갖고 asset 스케줄이 아닌 DAG는 일련의 구간들을 정의해요. 스케줄러는 이 구간들을 개별 Dag Run으로 바꿔 실행하죠. 기본적으로, DAG가 활성화될 때 마지막 데이터 구간 이후로 실행된 적이 없는 구간들은 스케줄러가 자동으로 만들지 않아요(Airflow 설정 scheduler.catchup_by_default=False). 스케줄러는 최신 구간에 대해서만 Dag Run을 만들어요.

DAG에서 catchup=True를 설정하면, 스케줄러는 마지막 데이터 구간 이후 실행된 적이 없거나(또는 clear된) 모든 구간에 대해 Dag Run을 만들어 실행해요. 이 개념이 바로 Catchup이에요.

DAG가 catchup에 대비해 작성돼 있지 않다면 (예컨대 구간에 맞추는 게 아니라 Now 시점에 맞춰 동작한다면) catchup을 꺼야 해요. 기본 설정이 그렇고, 환경에서 기본 설정이 바뀌었다면 DAG 정의에 catchup=False를 명시하면 돼요.

"""
Airflow 튜토리얼 예제:
https://github.com/apache/airflow/blob/main/airflow/example_dags/tutorial.py
"""

from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator

import datetime
import pendulum

dag = DAG(
    "tutorial",
    default_args={
        "depends_on_past": True,
        "retries": 1,
        "retry_delay": datetime.timedelta(minutes=3),
    },
    start_date=pendulum.datetime(2015, 12, 1, tz="UTC"),
    description="A simple tutorial Dag",
    schedule="@daily",
)

위 예시에서 DAG가 2016-01-02 오전 6시에 스케줄러 데몬(또는 명령줄)에 의해 잡힌다고 해볼게요. 그러면 데이터 구간이 2016-01-012016-01-02인 Dag Run이 하나 만들어지고, 다음 실행은 2016-01-03 자정 직후에 데이터 구간 2016-01-022016-01-03으로 생성돼요.

여기서 datetime.timedelta 객체를 schedule로 쓰면 동작이 달라질 수 있어요. 그 경우 만들어지는 단일 Dag Run은 2016-01-01 06:00 ~ 2016-01-02 06:00(지금을 끝으로 하는 스케줄 구간 하나)을 덮게 돼요. cron 기반과 delta 기반 스케줄의 차이가 궁금하다면 timetables comparison을 확인해 보세요.

만약 dag.catchup 값이 True였다면, 스케줄러는 2015-12-01과 2016-01-02 사이의 완료된 각 구간마다 Dag Run을 만들었을 거예요(아직 완료되지 않은 2016-01-02 구간은 제외). 그리고 그것들을 순차적으로 실행했을 거예요.

Catchup은 DAG를 일정 기간 꺼뒀다가 다시 켤 때도 발동돼요. 이 동작은 구간으로 쉽게 나눌 수 있는 원자적인 asset에 유용해요. 반대로 DAG가 catchup을 내부적으로 처리한다면 catchup을 꺼두는 게 좋고요.

Backfill

지정된 과거 기간 동안 DAG를 실행하고 싶을 수 있어요. 예를 들어 start_date가 2024-11-21인 DAG가 있는데, 다른 사용자가 한 달 전인 2024-10-21부터의 출력 데이터를 요구하는 상황이죠. 이런 과정을 Backfill이라고 불러요.

Backfill은 UI나 CLI로 실행할 수 있어요.

UI

Dag Details 페이지에서 Trigger를 클릭하고 Backfill을 선택하면 backfill 폼이 열려요. 날짜 범위, 재처리 동작, 최대 활성 실행 수(max active runs), 선택적인 역순 처리(run backwards), 고급 설정(Advanced Config)을 지정하면 돼요.

CLI

CLI로는 아래 명령어를 실행하면 돼요.

airflow backfill create --dag-id DAG_ID \
    --from-date START_DATE \
    --to-date END_DATE \
    --reprocess-behavior failed \
    --max-active-runs 3 \
    --run-backwards \
    --dag-run-conf '{"my": "param"}'

backfill 명령어는 시작 날짜와 종료 날짜 사이의 모든 구간에 대해 해당 dag_id의 인스턴스를 전부 다시 실행해요.

태스크 다시 실행하기 (Re-run Tasks)

스케줄 실행 중 일부 태스크가 실패할 수 있어요. 로그를 확인해 오류를 고친 뒤, 해당 날짜의 태스크 인스턴스를 clear해서 다시 실행할 수 있어요. 태스크 인스턴스를 clear하면 그 인스턴스의 기록이 남고, try_number가 증가하며 max_tries는 0, 상태는 None으로 바뀌어 태스크가 재실행돼요.

Airflow 3.1.0의 실험적 기능으로, 태스크 인스턴스를 clear하고 최신 번들 버전으로 다시 실행할 수도 있어요.

Tree나 Graph 뷰에서 실패한 태스크를 클릭한 뒤 Clear를 누르면 돼요. 그러면 실행기가 그 태스크를 다시 실행해요.

다시 실행할 때 선택할 수 있는 옵션이 여러 가지예요.

• Past - DAG의 가장 최근 데이터 구간 이전 실행들의 태스크 인스턴스 전체
• Future - 가장 최근 데이터 구간 이후 실행들의 태스크 인스턴스 전체
• Upstream - 현재 DAG의 업스트림 태스크
• Downstream - 현재 DAG의 다운스트림 태스크
• Recursive - 자식 DAG와 부모 DAG의 모든 태스크
• Failed - DAG의 가장 최근 실행에서 실패한 태스크만

CLI로도 clear할 수 있어요.

airflow tasks clear dag_id \
    --task-regex task_regex \
    --start-date START_DATE \
    --end-date END_DATE

지정된 dag_id와 시간 구간에 대해, 해당 정규식과 일치하는 태스크의 인스턴스를 전부 clear해요. 더 많은 옵션은 clear 명령어 도움말에서 확인할 수 있어요.

airflow tasks clear --help

태스크 인스턴스 기록 (Task Instance History)

태스크 인스턴스가 재시도되거나 clear되면 그 기록(history)이 보존돼요. Grid 뷰에서 해당 태스크 인스턴스를 클릭하면 이 기록을 볼 수 있어요.

위에 보이는 try 선택기는 재시도되거나 clear된 태스크에 대해서만 제공돼요.

기록에는 각 실행이 끝난 시점의 태스크 인스턴스 속성 값이 담겨요. 로그 페이지에서는 재시도마다의 로그도 볼 수 있어서 디버깅에 유용해요.

참고로 XCom, 렌더링된 템플릿 필드 같은 관련 객체는 기록에 보존되지 않아요. 로그를 포함한 태스크 인스턴스 속성만 보존돼요.

외부 트리거 (External Triggers)

Dag Run은 CLI를 통해서도 수동으로 만들 수 있어요. 아래 명령어를 실행하면 돼요.

airflow dags trigger --logical-date logical_date run_id

스케줄러 밖에서 만들어진 Dag Run은 트리거의 타임스탬프와 연결되고, UI에서 스케줄된 Dag Run과 함께 표시돼요. DAG 내부에 전달되는 논리적 날짜는 -e 인자로 지정할 수 있고, 기본값은 UTC 시간대의 현재 날짜예요.

웹 UI에서도 수동 트리거할 수 있어요(Dags 탭 → Links 열 → Trigger Dag 버튼).

DAG 트리거 시 파라미터 전달하기

CLI, REST API, UI에서 DAG를 트리거할 때 Dag Run 설정을 JSON blob으로 전달할 수 있어요.

파라미터화된 DAG 예시:

import pendulum

from airflow.sdk import DAG
from airflow.providers.standard.operators.bash import BashOperator

dag = DAG(
    "example_parameterized_dag",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
)

parameterized_task = BashOperator(
    task_id="parameterized_task",
    bash_command="echo \"here is the message: '$message'\"",
    env={"message": '{{ dag_run.conf["message"] if dag_run else "" }}'},
    dag=dag,
)

참고: dag_run.conf의 파라미터는 연산자의 템플릿 필드에서만 사용할 수 있어요.

Dag Run 기다리기 (Wait for a Dag Run)

Airflow는 Dag run이 완료될 때까지 기다리는 실험적 API를 제공해요. Airflow를 외부 시스템이나 자동화 파이프라인에 통합할 때, DAG가 끝날 때까지 실행을 멈추고 기다려야 한다면 특히 유용해요.

이 엔드포인트는 지정된 Dag run이 success, failed, canceled 같은 종료 상태에 도달할 때까지 블로킹(폴링)해요.

응답은 NDJSON(Newline-Delimited JSON) 형식으로 스트리밍돼요. 각 줄은 그 시점의 Dag run 상태를 나타내는 JSON 객체예요.

예를 들어:

{"state": "running"}
{"state": "success", "results": {"op": 42}}

클라이언트는 이걸로 실행을 실시간 감시하고, 특정 태스크의 XCom 결과를 선택적으로 수집할 수 있어요.

참고로 이 기능은 실험적이라 향후 Airflow 버전에서 바뀌거나 제거될 수 있어요.

CLI 사용법:

airflow dags trigger --conf '{"conf1": "value1"}' example_parameterized_dag

기억할 점

• 태스크 인스턴스를 UI에서 실패(failed)로 표시할 수 있어요. 실행 중인 태스크 인스턴스를 중단할 때 써요.
• 태스크 인스턴스를 성공(success)으로 표시할 수도 있어요. 주로 오탐(false negative)을 바로잡거나, 수정이 Airflow 밖에서 이뤄졌을 때 사용해요.

더 알아보기

• DAGs - DAG 기본 개념
• Scheduling - 스케줄링과 타임테이블
• Tasks - 태스크와 태스크 인스턴스
• Backfill - 과거 구간 실행