1. Airflow 3, Revisited

• 2.10.3 기준으로 써 둔 Airflow 노트를 3.3.2(stable) 기준으로 다시 정리한 글
• 원래 노트의 순서(Component, Executor, Callback, DB Connection)를 살리고, 3.0부터 3.3까지 새로 들어온 기능을 함께 다룸
• 각 절마다 2.x와 달라진 점을 따로 적음
• 같은 내용을 발표 덱으로 만든 버전은 2. Airflow 3 Slides

flowchart TD
  a3["Airflow 3"] --> arch["구조"]
  arch --> api["API Server + Task Execution API"]
  arch --> dp["DAG Processor 분리"]
  arch --> sdk["Task SDK (airflow.sdk)"]
  a3 --> input["코드와 데이터가 들어오는 길"]
  input --> bundle["Dag bundle, DAG 버저닝"]
  input --> asset["Asset, 이벤트 기반 스케줄링"]
  a3 --> sched["스케줄과 실행"]
  sched --> di["data_interval 기본값 변경"]
  sched --> exec["Executor (Edge, multi-executor)"]
  a3 --> ops["알림과 새 기능"]
  ops --> deadline["Deadline Alerts"]
  ops --> hitl["Human-in-the-Loop"]
  ops --> state["State Store"]

1.1 Release Timeline

3.x는 1년 반 동안 네 번 크게 바뀌었음. 3.0이 구조를 바꿨고, 3.1부터 3.3까지는 그 위에 기능을 쌓음. 우리 레포는 아직 2.10.3임.

현재 stable은 3.3.2임.

1.2 Three Big Changes

3.0의 큰 변화 세 가지는 모두 “워커가 무엇에 직접 닿을 수 있는가"를 다시 정한 결과임.

1. Webserver가 API Server로

• UI, REST API /api/v2, 워커용 Task Execution API를 서버 하나가 맡음
• FastAPI 기반
• 기동 명령: airflow api-server

2. DAG Processor는 따로 띄움

• 2.x에서는 scheduler 안의 subprocess였음
• 3.x에서는 반드시 따로 띄우는 독립 컴포넌트
• 기동 명령: airflow dag-processor

3. 워커의 metadata DB 직접 접근 차단

• 상태 전이, heartbeat, XCom, Variable과 Connection 조회가 전부 API Server를 거침
• 워커 쪽 인터페이스는 airflow.sdk(Task SDK)
• 영향: task 코드에서 DB session을 쓸 수 없음

1.3 Architecture: 2.x vs 3.x

2.x: 모든 컴포넌트가 DB에 직접 붙음

flowchart LR
  files["DAG files<br/>dags/ 폴더"] --> proc["DagFileProcessor<br/>(scheduler 안의 subprocess)"]
  proc --> db[("Metadata DB")]
  proc -.-> sched["Scheduler + Executor"]
  sched <--> db
  sched -- "실행 요청" --> worker["Worker"]
  worker -- "DB 직접 접속" --> db
  trig["Triggerer"] <--> db
  web["Webserver<br/>Flask, /api/v1"] --> db

• 워커까지 metadata DB에 직접 붙기 때문에 워커 수만큼 DB 커넥션이 늘어남
• DagFileProcessor는 독립 프로세스가 아니라 scheduler가 띄우는 subprocess

3.x: 워커와 DB 사이에 API Server가 들어감

flowchart LR
  files["DAG files<br/>Dag bundle"] --> proc["DAG Processor<br/>파싱, DAG callback"]
  proc -- "serialized DAG" --> db[("Metadata DB")]
  sched["Scheduler + Executor"] <--> db
  sched -- "실행 요청" --> worker["Worker"]
  worker -- "Task SDK<br/>상태, XCom, Connection" --> api["API Server<br/>Execution API, UI"]
  api --> db
  client["UI, REST client<br/>브라우저, airflowctl"] -- "/api/v2" --> api
  trig["Triggerer"] <--> db

• 워커는 API Server와만 통신함
• 단, DAG Processor와 Triggerer에서 도는 Dag 작성자 코드는 3.x에서도 DB에 접근할 수 있음(공식 security model)

1.4 Components

역할이 그대로인 것과 바뀐 것을 나눠 보면 정리가 쉬움.

1.5 Task SDK

AIP-72, AIP-108. 워커가 API로만 통신하게 되면서 task를 실행하는 쪽이 Airflow core와 분리됨. 이 분리 덕분에 다른 언어로 task를 쓸 수도 있게 됨.

3.0부터 3.2까지: 분리

• 3.0: Task Execution API와 Task SDK 도입
• 3.1: 직렬화 개선, 예전 import 경로에 경고
• 3.2: 예외, serde, lineage도 SDK로 이동
• 결과적으로 SDK만 따로 업그레이드할 수 있게 됨

3.3: Java와 Go (experimental)

• 오케스트레이션(DAG 정의)은 Python 그대로
• task 본문만 Java나 Go로 쓰고 일반 워커에서 실행함
• XCom, Connection, Variable은 SDK client로 읽음
• 언어별 SDK는 pip 패키지와 별도로 배포됨

# Python 쪽은 task 모양만 선언하고, queue로 언어별 coordinator에 연결한다
@task.stub(queue="golang")
def go_transform(): ...

@task.stub(queue="java")
def java_load(): ...

# airflow.cfg
[sdk]
queue_to_coordinator = {"golang": "go", "java": "jdk-17"}

• dag_id와 task_id는 Python stub과 Go, Java 코드에서 똑같아야 함

1.6 Dag Bundles and DAG Versioning

AIP-66. 2.x에서는 run 도중 DAG 파일이 바뀌면 남은 task가 새 코드로 돌 수 있었음. 3.x는 버저닝되는 bundle이면 run이 시작한 버전을 끝까지 씀.

gitGraph
  commit id: "v1 (a1b2c3)"
  commit id: "run 10-18 시작: extract v1"
  commit id: "v2 머지 (task 이름 변경)" type: HIGHLIGHT
  commit id: "run 10-18은 transform v1, load v1로 끝까지"
  commit id: "run 10-19: extract v2, clean v2"

• UI에서 run마다 그때의 DAG 구조를 볼 수 있음

Bundle 종류

• [dag_processor] dag_bundle_config_list로 여러 개를 등록함
• 지금처럼 git-sync 사이드카로 폴더에 동기화하면 LocalDagBundle이라 버전이 고정되지 않음

clear, rerun, backfill이 쓰는 버전 (3.3)

• 기본값: clear와 rerun은 원래 버전, backfill은 최신 버전
• DAG의 rerun_with_latest_version이나 [core] 설정으로 바꿈
• backfill은 3.0부터 scheduler가 관리하고 UI에서 만듦

1.7 Assets

AIP-74, 75, 76. 2.x의 Dataset이 Asset이 됨. 시간이 아니라 데이터 갱신으로 DAG를 돌리는 기능.

3.0: @asset과 이벤트 기반 스케줄링

• airflow.datasets.Dataset은 airflow.sdk.Asset으로 바뀜
• @asset: 함수 하나가 asset을 만드는 DAG가 됨
• AssetWatcher: 메시지 큐 같은 외부 이벤트로 asset을 갱신함
• watcher의 trigger는 BaseEventTrigger를 상속해야 함

주의: 무한 스케줄링

• “파일이 있다”, “작업이 성공했다"처럼 한 번 참이면 계속 참인 조건으로 watcher를 걸면 DAG가 반복해서 돎
• 예: S3KeyTrigger

3.2: Asset 파티셔닝

같은 파티션이 모두 찼을 때만 downstream이 돎.

• upstream은 CronPartitionTimetable로 시간 파티션을 씀
• downstream은 PartitionedAssetTimetable(assets=a & b)로 같은 파티션 키가 둘 다 찼을 때만 돎
• 3.3에서 FanOutMapper, WaitForAll, MinimumCount(n) 같은 정책이 추가됨

1.8 Deferrable Operators

기다리는 동안에는 워커 슬롯을 반납함. operator 전체가 triggerer에서 도는 게 아님. 시작과 재개는 워커에서, 기다림만 triggerer에서 처리함.

sequenceDiagram
  participant W as Worker slot
  participant T as Triggerer
  W->>W: execute() (running, 슬롯 점유)
  W->>T: self.defer()
  Note over W: 슬롯 반납, 그동안 다른 task가 씀
  T->>T: async trigger 폴링 (deferred)
  T->>W: TriggerEvent
  W->>W: resume (running)

1.9 data_interval: 가장 먼저 다시 볼 것

같은 cron인데 data_interval의 모양이 다름. 원래 노트의 “2시간 스케줄, 자정 실행이면 22:00부터 00:00까지” 예시는 3.x 기본 설정에서 더 이상 맞지 않음.

• 3.x 기본값 create_cron_data_intervals = False
• 그래서 cron 문자열에는 CronTriggerTimetable이 붙음

예시: schedule="0 */2 * * *", 2026-10-18 00:00에 도는 run

• 2.x: run은 구간(22:00부터 00:00까지)이 끝난 뒤 00:00에 돌고, logical_date는 구간 시작
• 3.x: data_interval이 폭 0인 점이고, logical_date는 트리거 시각과 같음
• ds가 2.x보다 하루 뒤라서 “전날 데이터"를 ds로 계산하던 로직은 어긋남

'@daily'도 같은 규칙

• 원래 노트에 “확인 필요"로 남겼던 부분
• apache/airflow 소스 task-sdk/src/airflow/sdk/definitions/dag.py의 _create_timetable에서 확인함
• '@daily' 같은 preset도 문자열이라 같은 분기를 탐
  • create_cron_data_intervals가 True면 CronDataIntervalTimetable
  • False(3.x 기본값)면 CronTriggerTimetable
• timedelta 스케줄은 같은 설정에 따라 DeltaDataIntervalTimetable이나 DeltaTriggerTimetable이 됨
• ds는 UTC logical_date를 strftime("%Y-%m-%d")한 값

if isinstance(interval, str):
    if airflow_conf.getboolean("scheduler", "create_cron_data_intervals"):
        return CronDataIntervalTimetable(interval, timezone)
    return CronTriggerTimetable(interval, timezone=timezone)

구간이 필요하면 업그레이드 전에

• create_cron_data_intervals = True를 업그레이드 전에 설정하거나 CronDataIntervalTimetable을 명시함
• 3.x run이 쌓인 뒤에 바꾸면 스케줄 run 하나가 skip됨

1.10 logical_date in Manual Runs

수동 run에서는 logical_date를 직접 읽음.

바뀐 점

• REST API로 trigger하면 logical_date 기본값이 None
• 수동 run의 data_interval은 timetable과 trigger 경로에 따라 달라짐
• execution_date, next_ds, prev_ds, tomorrow_ds, yesterday_ds는 context에서 빠짐

점검할 곳

• 수동 trigger 때 data_interval_start를 “요청 날짜"로 쓰던 DAG
• TriggerDagRunOperator로 하위 DAG를 띄우는 흐름
• DT Platform Manual Trigger처럼 API로 run을 만드는 기능

# 사용자가 지정한 날짜가 필요할 때
from airflow.sdk import get_current_context, task

@task
def process():
    ctx = get_current_context()
    target = ctx["logical_date"]          # 수동 run의 요청 날짜
    window = (ctx["data_interval_start"], ctx["data_interval_end"])
    # 구간 의미가 필요할 때만 data_interval을 쓴다

1.11 Executors

Celery와 Kubernetes는 개념이 2.x와 같음. 다른 점은 워커가 DB 대신 API Server와 통신한다는 것. API로만 통신하니 원격지 워커(Edge)도 가능해짐.

CeleryKubernetesExecutor 대신 multi-executor

[core]
executor = CeleryExecutor,KubernetesExecutor   # 첫 번째가 기본값

heavy = PythonOperator(
    task_id="heavy",
    python_callable=run,
    executor="KubernetesExecutor",
)

• 3.2 multi-team(experimental)에서는 팀마다 다른 executor, connection, pool을 쓸 수 있음

1.12 Callbacks

콜백은 워커 실행으로 상태가 바뀔 때만 불림. CLI나 UI로 상태를 바꾸면 불리지 않음.

Task-level: operator나 default_args

• 실행 위치: Worker, 에러는 task 로그에 남음

flowchart LR
  s["scheduled"] --> q["queued"] --> ex(["on_execute_callback"]) --> r["running"]
  r --> ok["success"] --> cbok(["on_success_callback"])
  r --> f["failed"] --> cbf(["on_failure_callback"])
  r --> rt["up_for_retry"] --> cbr(["on_retry_callback"])
  r --> sk["skipped"] --> cbs(["on_skipped_callback"])

DAG-level: DAG(…)

• 실행 위치: DAG Processor, 에러는 dag processor 로그에 남음

flowchart LR
  run["DagRun running"] --> ok["success"] --> cbok(["on_success_callback"])
  run --> f["failed"] --> cbf(["on_failure_callback"])

• 콜백은 리스트로도 지정함: on_failure_callback=[f1, f2]
• 3.2부터 DAG 콜백 context의 TI는 DAG 상태와 관련된 task로 골라짐

1.13 From SLA to Deadline Alerts

AIP-86. SLA는 끝난 task만 판정해서 막힌 run을 못 잡았음.

SLA (3.0에서 제거)

• task가 끝나야만 판정함. 막혀 있는 task는 miss를 못 잡음
• SLA는 task 속성인데 callback은 DAG 속성임
• 판정할 때마다 DAG를 다시 파싱함

Deadline Alerts (3.1부터)

• 기준 시각(큐 진입, logical_date, 고정 시각)에 허용 시간을 더해 판정함
• 3.1: triggerer에서 도는 AsyncCallback
• 3.2: 워커에서 도는 SyncCallback, DAG당 여러 개
• 3.3: Browse 메뉴에 Deadlines 페이지

우리 DAG의 sla_miss_callback 옮기기

from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference, AsyncCallback

with DAG(
    "datamodeling_temp_layer_daily_kr",
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_QUEUED_AT,   # 큐에 들어간 시각부터
        interval=timedelta(hours=2),                    # 2시간 안에 안 끝나면
        callback=AsyncCallback(SlackWebhookNotifier, kwargs={...}),
    ),
): ...

• 기존 sla_miss_callback=SLAMissNotifier()의 자리
• 3.2 기능은 experimental이라 버전별 문서를 같이 봄

1.14 New in 3.1 to 3.3

task가 사람을 기다리고, 상태를 기억함.

Human-in-the-Loop (3.1)

• HITLOperator가 deferred 상태로 멈추고 UI에 입력 폼을 띄움
• 권한 있는 사용자가 params와 XCom을 보고 결정함
• 3.3: awaiting_input 상태 추가
• 예: 모델 결과 검수, 적재 전 승인

State Store (3.3)

• retry와 run을 넘어 남는 key-value
• task_state_store: task instance 범위
• asset_state_store: asset 범위, 공유 watermark
• Spark 재시도 때 기존 job에 다시 붙음(durable=True)
• 예: 증분 적재 cursor 저장

@task
def extract(**context):
    state = context["task_state_store"]
    cursor = state.get("last_cursor", default=0)   # 지난 run이 남긴 값
    rows = fetch_records(since=cursor)
    state.set("last_cursor", rows[-1]["id"])
    return rows

그 밖에

• 3.3 Pluggable Retry Policy: 예외 종류별 재시도와 backoff
  • 공식 블로그에 API 예시가 없어 개념만 적음
• 3.2 async def를 PythonOperator와 @task에 쓸 수 있음
• 3.1 trigger rule ALL_DONE_MIN_ONE_SUCCESS
• 3.3 @result와 /wait API로 run 결과 받기(experimental)

1.15 DB Connections from Tasks and Callbacks

외부 DB와 Airflow metadata DB는 붙는 길이 다름.

• 콜백은 상태 변화에 반응해 알리는 용도
• 안에서 무거운 query나 대량 insert, update를 하면 장애 지점과 운영 부담이 커짐
• dag_run을 BQ로 싱크하는 notifier처럼 metadata를 읽는 코드는 3.x로 옮길 때 점검함
• 참고: data-infrastructure 레포의 alpha airflow rds-proxy.tf (Andy 공유)

1.16 UI and Operations

UI는 React로, timer 메트릭은 Histogram으로 바뀜.

UI

• React로 새로 쓴 UI, Grid 안에 Gantt
• 3.1: UI 17개 언어 지원, 즐겨찾기 DAG
• 3.2: UI에서 XCom 추가, 수정, 삭제
• 3.3: DagRun 일괄 mark, TI 일괄 clear

API, CLI, 플러그인

• REST /api/v2, 원격 CLI airflowctl 분리
• Flask 플러그인 대신 React 앱, FastAPI 앱, 미들웨어
• 기존 FAB 뷰는 FAB provider로 호환됨

메트릭 주의 (3.3)

• timer 메트릭이 OTel Gauge에서 Histogram으로 바뀜
• dag_processing.last_run.seconds_ago가 tag 방식으로 바뀜
• 기존 대시보드와 알림 쿼리를 고쳐야 함

1.17 Migrating Our DAG

datamodeling_temp_layer_daily_kr 기준. 바뀌는 곳은 import, schedule, SLA 세 군데.

- from airflow.models import DAG
+ from airflow.sdk import DAG

  with DAG(
      dag_id=DAG_ID,
-     schedule_interval='@daily',
+     schedule='@daily',
      start_date=datetime.datetime(2024, 4, 7, 0, tzinfo=KST),
+     catchup=False,                      # 3.x 기본값도 False지만 명시한다
      default_args=default_args,          # FailureNotifier, ResolveSlackAlertNotifier 그대로
      max_active_runs=1,
      max_active_tasks=20,
-     sla_miss_callback=SLAMissNotifier(),
+     # Deadline Alerts로 이전 (1.13)
      on_success_callback=dag_run_sync_to_bq_notifier,
      on_failure_callback=dag_run_sync_to_bq_notifier,
  )

• 커스텀 Notifier는 airflow.sdk.BaseNotifier를 상속하게 바꿈
• '@daily'에서 ds를 쓰는 로직은 1.9 기준으로 확인함

1.18 Checklist

import 경로

바뀐 기본값과 제거된 것

• schedule_interval, timetable 인자는 schedule 하나로
• schedule 기본값은 None, catchup 기본값은 False
• xcom_pull()은 task_ids가 없으면 현재 task만 봄
• SubDAG는 TaskGroup이나 Asset으로
• REST /api/v1은 /api/v2로

자동 점검

ruff check dags/ --select AIR301
ruff check dags/ --select AIR301 --fix

# 설정 파일 갱신
airflow config update

핵심 정리

1. 워커는 metadata DB에 직접 닿지 않음. 모든 런타임 상호작용은 API Server의 Task Execution API를 거침
2. DAG Processor는 따로 띄우고, DAG 파일은 Dag bundle로 들어옴. git bundle이면 run마다 버전이 고정됨
3. cron 문자열('@daily' 같은 preset 포함)의 기본 timetable이 CronTriggerTimetable이라 data_interval과 ds가 달라짐
4. 시간 대신 Asset으로 돌릴 수 있고, 3.2부터는 파티션 단위로 깨움
5. SLA는 Deadline Alerts로, 사람 승인은 HITL로, 증분 cursor는 State Store로 처리함

출처

• Upgrading to Airflow 3
• Release Notes (3.3.2)
• Apache Airflow 3.1.0 블로그
• Apache Airflow 3.2.0 블로그
• Apache Airflow 3.3.0 블로그
• Dag Bundles
• Event-driven scheduling
• Callbacks
• apache/airflow 소스: task-sdk/src/airflow/sdk/definitions/dag.py (_create_timetable)