Airflow

Step 5: 커스텀 태스크의 DB 직접 접근 제거하기

Airflow 3에서는 오퍼레이터가 DB 세션을 이용해 메타데이터 DB에 직접 접근할 수 없다. 커스텀 오퍼레이터가 있다면 코드에 DB 직접 접근 호출이 없는지 검토해야 한다. 참고할 실제 사례는 GitHub 이슈 #49187에 정리되어 있다.

과거 태스크 코드가 메타데이터 DB에 직접 접근했다면, 다음 두 접근법 중 하나로 이전해야 한다.

권장 접근법: Airflow Python Client

공식 Airflow Python Client를 사용해 REST API를 통해 메타데이터와 상호작용하는 것이 권장되는 방법이다. 이 클라이언트는 DagRun, TaskInstance, Variable, Connection, XCom 등 대부분의 사용 사례에 대한 API를 제공한다.

장점은 워커에서 DB로의 직접 네트워크 접근이 필요 없다는 점, Airflow 3의 API-first 아키텍처와 가장 잘 부합한다는 점, 워커 환경에 DB 자격증명이 필요 없이 API 토큰만 있으면 된다는 점, 워커에 DB 드라이버 설치가 필요 없다는 점, 그리고 API 서버를 통해 접근 제어와 인증이 중앙화된다는 점이다.

단점은 `apache-airflow-client` 패키지를 별도로 설치해야 한다는 점, `/auth/token` API 호출로 액세스 토큰을 받고 필요에 따라 로테이션해야 한다는 점, API 서버 가용성과 네트워크 접근이 필요하다는 점, 그리고 모든 DB 연산이 API 엔드포인트로 노출되어 있지는 않을 수 있다는 점이다. 필요한 기능이 API로 제공되지 않는다면, 직접 DB 접근을 시도하기보다 새 API 엔드포인트나 Task SDK 기능을 요청하는 것을 커뮤니티는 우선시한다.

권장되지 않는 워크어라운드: DbApiHook

Python Client를 쓸 수 없는 사용자를 위한 알려진 워크어라운드로 `PostgresHook`/`MySqlHook` 같은 `DbApiHook`을 이용해 메타데이터 DB에 직접 쿼리하는 방법이 문서화되어 있다. 다만 이 방법은 명시적으로 '권장되지 않음(NOT recommended)'이며, 심각한 한계가 있고 향후 Airflow 버전에서 반드시(will) 깨질 것이라고 경고된다.

구체적 위험은 다음과 같다: Airflow 3.2 이후 이 접근은 깨질 것이며 스키마 변경에 따른 대응은 전적으로 사용자 책임이다. 메타데이터 DB 스키마는 공개 API가 아니므로 사전 통지 없이 언제든 바뀔 수 있어 쿼리가 예고 없이 깨질 수 있다. 이 방식은 Airflow 3의 핵심 기능인 태스크 격리 원칙에 정면으로 위배된다. 또한 태스크마다 별도 DB 커넥션을 여는 Airflow 2 시절의 동작을 다시 불러들여 성능/확장성 특성을 크게 바꿔버린다.

Python Client로 해결할 수 없고 이 위험을 이해한 사용자만 메타데이터 DB(PostgreSQL 또는 MySQL, 실제 메타 DB 타입과 일치)에 연결하는 Connection을 만들고 Database Hook을 사용할 수 있다. 이 훅들은 API 서버를 경유하지 않고 psycopg2, mysqlclient 같은 DB 드라이버로 직접 연결한다는 점에 유의해야 한다.

from airflow.sdk import task
from airflow.providers.postgres.hooks.postgres import PostgresHook


@task
def get_connections_from_db():
    hook = PostgresHook(postgres_conn_id="metadata_postgres")
    records = hook.get_records(sql="""
        SELECT conn_id, conn_type, host, schema, login
        FROM connection
        WHERE conn_type = 'postgres'
        LIMIT 10;
        """)

    return records

오퍼레이터 스타일을 선호한다면 `SQLExecuteQueryOperator`도 같은 목적으로 사용할 수 있다. 이 방식을 쓰더라도 메타데이터 DB 연결에는 항상 읽기 전용(read-only) 자격증명을, 가능하면 임시 자격증명을 사용하는 것이 권장된다.

Step 6: 배포 관리자의 인스턴스 업그레이드

더 쉽고 안전한 업그레이드를 위해 인스턴스 설정을 점검하는 유틸리티가 제공된다. 먼저 아래 명령으로 설정을 점검한다.

airflow config update

이 유틸리티는 설정을 Airflow 3와 자동으로 호환되도록 갱신할 수도 있다.

airflow config update --fix

업그레이드에서 가장 큰 부분은 DB 업그레이드이며, Airflow 3의 DB 업그레이드 과정은 Airflow 2.7 이상과 동일하게 `airflow db migrate`로 수행한다.

Flask-AppBuilder 뷰(`appbuilder_views`), FAB 메뉴 아이템(`appbuilder_menu_items`), Flask 블루프린트(`flask_blueprints`)를 사용하는 플러그인이 있다면, 이를 FastAPI 앱으로 전환하거나 하위호환 레이어를 제공하는 FAB provider를 설치해야 한다. 이상적으로는 플러그인을 Airflow 3 플러그인 인터페이스(External Views `external_views`, FastAPI 앱 `fastapi_apps`, FastAPI 미들웨어 `fastapi_root_middlewares`)로 전환하는 것이 권장된다.

Helm 차트로 배포한다면 `values.yaml`에 정의된 값들을 Airflow 3에서 가능한 설정 옵션과 대조해 확인해야 한다. `webserver` 아래에 있던 모든 설정은 `apiServer`로 이름이 바뀌어야 하며, 많은 파라미터가 이름이 바뀌거나 제거되었다는 점을 고려해야 한다. 차트별 전체 업그레이드 체크리스트(values.yaml 변경, 독립 실행형 Dag processor, JWT secret, FAB 기본값, 최소 Kubernetes 버전, 차트 1.16.0~1.18.0 사이의 이름 변경 키 등)는 `helm-chart:upgrading-to-airflow-3` 문서를 참고한다.

Step 7~8: 시작 스크립트 변경과 업그레이드 후 점검

Airflow 3에서 Webserver는 범용 API 서버가 되었다. API 서버는 다음 명령으로 시작한다.

airflow api-server

Dag 프로세서는 이제 로컬/개발 환경에서도 반드시 독립적으로 시작해야 한다.

airflow dag-processor

업그레이드 후 확인해야 할 대표적 사항은 SSO(OAuth, OIDC, LDAP) 인증이 예상대로 동작하는지다. 커스텀 `webserver_config.py`를 사용한다면 `from airflow.www.security import AirflowSecurityManager`를 `from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride`로 교체해야 한다.