Airflow

Airflow 2.x 아키텍처: 컴포넌트의 DB 직접 접근

Airflow 2.x에서는 모든 컴포넌트가 Airflow 메타데이터 DB와 직접 통신했다. 태스크 코드와 이를 실행하는 airflow 패키지 코드가 같은 프로세스, 같은 네트워크 공간 안에서 함께 실행되도록 설계되어 있었다. 이 구조에서 워커는 DB에 직접 연결해 사용자 코드를 실행했는데, 이는 사용자 코드가 세션을 임포트해 메타데이터 DB에 악의적인 작업을 수행할 수 있는 구조적 취약점이었다. 또한 컴포넌트마다 DB에 직접 연결하다 보니 연결 수가 과도해져 스케일링 문제로 이어졌다.

Airflow 3.x 아키텍처: API 서버를 경유하는 실행 구조

Airflow 3.x에서는 API 서버가 태스크/워커를 위한 메타데이터 DB의 유일한 접근 지점이 된다. 이 API 서버는 여러 애플리케이션을 함께 지원한다 — Airflow REST API, 정적 JS를 서빙하는 Airflow UI용 내부 API, 그리고 워커가 태스크 실행 인터페이스를 통해 TI(TaskInstance)를 실행할 때 상호작용하는 API. 워커는 더 이상 DB와 직접 통신하지 않고 API 서버와 통신한다.

Dag 프로세서와 트리거러(Triggerer) 역시 변수나 커넥션이 필요할 때 이 태스크 실행 메커니즘을 활용한다.

이에 따라 Airflow 3에서는 태스크 코드에서 메타데이터 DB로의 직접 접근이 제한된다. 태스크 코드는 더 이상 Airflow DB 세션이나 모델을 직접 임포트해 쓸 수 없으며, 상태 전이·하트비트·XCom·리소스 조회 같은 모든 런타임 상호작용은 전용 Task Execution API를 통해 이루어진다. 이는 워커 태스크 코드가 메타데이터 DB에 직접 접근/수정하는 것을 막아 격리와 보안을 강화한다. Task SDK는 DB에 직접 의존하지 않고 Airflow 리소스에 접근하는 안정적이고 하위호환 가능한 인터페이스를 제공한다. 다만 Dag 작성자의 코드가 Dag File Processor와 Triggerer 안에서는 여전히 직접 DB 접근을 가지고 실행될 수 있다는 점은 `/security/security_model` 문서에서 별도로 다룬다.

Step 1~2: 사전 준비, 백업과 정리

업그레이드 전 준비 단계로 먼저 Airflow 2.7 이상인지 확인해야 한다(최신 2.x로 먼저 올린 뒤 3으로 가는 것이 권장된다). Python 버전이 지원 목록에 있는지, 그리고 Airflow 3에서 제거된 기능/동작을 사용하고 있지 않은지도 확인해야 한다.

다음으로 메타데이터 DB를 반드시 백업해야 한다. 핫 백업이 불가능하다면 인스턴스를 셧다운한 뒤 백업해야 일관성이 보장된다(예: 인스턴스를 끄지 않으면 백업에 모든 TaskInstance/DagRun이 포함되지 않을 수 있다). 백업 없이 마이그레이션이 실패하면 절반만 마이그레이션된 상태에 빠질 수 있다(네트워크 단절 등이 원인이 될 수 있음).

오래 운영된 인스턴스는 더 이상 필요 없는 데이터(예: 오래된 XCom)가 많이 쌓여 있을 수 있다. Airflow 3 업그레이드에는 스키마 변경이 포함되므로, DB가 크면 스키마 변경에 시간이 오래 걸릴 수 있다. 더 빠르고 안전한 마이그레이션을 위해 업그레이드 전 `airflow db clean` CLI 명령으로 메타 DB를 정리하는 것이 권장된다.

또한 `AirflowDagDuplicatedIdException` 같은 Dag 처리 오류가 없는지 확인해야 한다. `airflow dags reserialize`가 오류 없이 실행되어야 하며, Dag 처리 오류가 있다면 업그레이드 전에 기존 인스턴스에 수정 사항을 배포하고 모든 Dag이 재처리되어(오류가 사라질 때까지) 기다린 후 업그레이드를 진행해야 한다.

Step 3: ruff 기반 Dag 호환성 점검 (AIR301/302/311/312)

업그레이드 마찰을 줄이기 위해 Airflow는 Ruff와 AIR 규칙을 결합한 Dag 업그레이드 점검 도구를 제공한다. AIR301과 AIR302는 Airflow 3의 파괴적 변경을 가리키고, AIR311과 AIR312는 당장 파괴적이지는 않지만 갱신이 강력히 권장되는 변경을 가리킨다.

최신 ruff 버전이 가장 최신 규칙을 갖지만, 최소 0.13.1 버전 이상을 사용해야 한다. 점검 명령:

ruff check dags/ --select AIR301

권장 수정 사항을 미리 보려면:

ruff check dags/ --select AIR301 --show-fixes

일부 변경은 자동 수정이 가능하다:

ruff check dags/ --select AIR301 --fix

일부 수정은 'unsafe'로 표시된다. Unsafe 수정은 보통 Dag 코드를 깨뜨리지는 않지만 런타임 동작을 일부 바꿀 수 있어 unsafe로 분류된다. 이를 적용하려면:

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

AIR 규칙에서 unsafe 수정은 임포트된 멤버의 이름은 그대로 두고 import 경로만 바꾸는 경우다 (예: `from airflow.sensors.base_sensor_operator import BaseSensorOperator`를 `from airflow.sdk.bases.sensor import BaseSensorOperator`로 바꾸는 경우 — 기존 import를 먼저 제거해야 하므로 unsafe). 반대로 safe 수정은 멤버 이름과 import 경로가 함께 바뀌는 경우다 (예: `from airflow.datasets import Dataset`를 `from airflow.sdk import Asset`로). 사용하지 않는 레거시 import를 제거하려면 `unused-import`(F401) 규칙을 별도로 활성화해야 한다.

주요 Import 경로 변경과 마이그레이션 타임라인

ruff가 많은 import 문제를 자동으로 고쳐주지만, Dag과 관련 코드가 Airflow 3에서 올바르게 임포트되도록 알아야 할 핵심 변경들이 있다. 예전 경로는 deprecated 상태이며 향후 버전에서 제거될 예정이다. 대표적인 매핑은 다음과 같다: `airflow.decorators.dag/task/task_group/setup/teardown` → `airflow.sdk.*`, `airflow.models.dag.DAG` → `airflow.sdk.DAG`, `airflow.models.baseoperator.BaseOperator` → `airflow.sdk.BaseOperator`, `airflow.sensors.base.BaseSensorOperator` → `airflow.sdk.BaseSensorOperator`, `airflow.hooks.base.BaseHook` → `airflow.sdk.BaseHook`, `airflow.datasets.Dataset/DatasetAlias/DatasetAll/DatasetAny` → `airflow.sdk.Asset/AssetAlias/AssetAll/AssetAny`, `airflow.models.connection.Connection` → `airflow.sdk.Connection`, `airflow.models.variable.Variable` → `airflow.sdk.Variable`, `airflow.io.*` → `airflow.sdk.io.*` 등이다.

마이그레이션 타임라인은 Airflow 3.1에서는 레거시 import가 deprecation 경고를 내면서도 계속 동작하고, 향후 버전(Future Airflow version)에서 완전히 제거될 예정이라는 것이다.

Step 4: Standard Provider 설치

`BashOperator`, `PythonOperator`, `ExternalTaskSensor`, `FileSensor` 등 흔히 쓰이던 Operator/Sensor/Trigger들은 과거 `airflow-core` 패키지에 번들되어 있었지만, 이제 별도 패키지인 `apache-airflow-providers-standard`로 분리되었다. 이 패키지는 Airflow 2.x에도 설치할 수 있어, 3으로 넘어가기 전에 미리 Dag의 참조를 standard provider 패키지 쪽으로 수정해둘 수 있다.