배포와 테스트: DAG validation, local test, CI, rollback

DAG 배포는 Python 파일 복사가 아니라 운영 변경이다

Airflow에서 DAG는 코드다. 그래서 배포도 단순히 dags/ 폴더에 Python 파일을 밀어 넣는 일이 아니라, 운영 시스템의 실행 계획을 바꾸는 변경으로 봐야 한다. 작은 오타 하나는 DAG import error가 되고, import 단계에서 외부 API를 호출하는 코드는 scheduler를 느리게 만들며, 실행 중인 DagRun 사이에 코드가 바뀌면 같은 run 안에서도 앞 task와 뒤 task가 서로 다른 가정을 할 수 있다.

좋은 배포 체계는 “배포 후 Airflow UI에서 빨간색을 보면 고친다”가 아니다. 배포 전에 최소한 다음 네 가지를 통과시킨다.

1. DAG validation: import error, 구조, 팀 규칙을 자동 검사한다.
2. Local test: 개발자가 로컬에서 한 run의 실행 흐름을 빠르게 확인한다.
3. CI gate: main 브랜치로 들어가기 전 동일한 검사를 반복 가능하게 만든다.
4. Rollback plan: 배포가 잘못됐을 때 어떤 버전으로, 어떤 범위까지 되돌릴지 정한다.

이 장의 핵심은 테스트 종류를 많이 나열하는 것이 아니라, Airflow DAG 변경을 검증 가능한 릴리스 단위로 만드는 것이다.

1. 배포 파이프라인의 안전 경계

운영자가 원하는 흐름은 단순하다.

각 단계의 실패 의미가 달라야 한다.

중요한 점은 “실행 테스트”가 모든 문제를 잡는다고 기대하지 않는 것이다. Airflow DAG는 외부 DB, object storage, Spark, API, 권한, 네트워크에 의존한다. 그래서 배포 체계는 얕고 빠른 검사와 느리지만 실제에 가까운 검사를 분리해야 한다.

2. DAG loader test: 가장 먼저 막아야 할 import error

Airflow 공식 문서는 DAG를 production-level code로 다루고 여러 테스트를 붙이라고 설명한다. 그중 가장 기본은 DAG loader test다. DAG 파일을 그냥 Python으로 실행했을 때 오류가 없어야 scheduler도 안전하게 parse할 수 있다.

python dags/customer_daily_sales.py

이 방식은 별도 테스트 코드를 작성하지 않아도 다음 문제를 빨리 발견한다.

• 문법 오류
• 설치되지 않은 Python 패키지
• 잘못된 import 경로
• DAG parse 시점에 실행되는 top-level code 오류
• 필수 환경변수 누락

단, 이 검사는 반드시 scheduler와 같은 의존성, 같은 환경변수, 같은 공통 코드에서 실행해야 의미가 있다. 개발자 노트북에서는 통과하지만 scheduler image에는 패키지가 없으면 운영 import error는 그대로 발생한다.

2.1 DagBag 기반 import error 테스트

CI에서는 개별 파일을 하나씩 실행하기보다 DagBag을 만들어 전체 DAG import error를 검사하는 패턴이 일반적이다.

# tests/test_dag_imports.py
from airflow.models import DagBag


def test_no_import_errors():
    dagbag = DagBag(dag_folder="dags", include_examples=False)
    assert dagbag.import_errors == {}

Airflow 3.x 문서 예시는 import 경로가 일부 달라질 수 있다. 중요한 것은 특정 import 경로가 아니라 DagBag이 실제 DAG folder를 읽고 .import_errors를 비워야 한다는 원칙이다.

2.2 top-level code를 테스트로 드러내기

DAG 파일의 top-level code는 scheduler가 DAG를 parse할 때마다 실행된다. 공식 문서는 DB 접근, network call, 무거운 연산, expensive import를 top-level에 두지 말라고 경고한다. 이런 코드는 기능상 맞아 보여도 scheduler 성능을 망가뜨린다.

나쁜 예시는 다음과 같다.

# 나쁨: DAG parse 때마다 외부 시스템 호출
config = requests.get("https://config.internal/pipeline").json()

with DAG(...):
    ...

권장 형태는 외부 호출을 task 실행 시점으로 밀어 넣는 것이다.

# 좋음: DAG parse는 가볍게 유지
with DAG(...):
    @task
    def load_config():
        return requests.get("https://config.internal/pipeline").json()

CI에서 DAG import 시간을 측정해 “한 파일 parse가 몇 초 이상이면 실패” 같은 guardrail을 둘 수도 있다. 엄격한 threshold는 팀 환경마다 다르지만, top-level network call이 들어오면 즉시 티가 난다.

3. DAG validation: 팀 규칙을 코드로 고정한다

DAG가 import된다고 해서 운영에 안전한 것은 아니다. 다음 같은 규칙은 리뷰어 기억에 맡기면 빠진다.

• 모든 DAG에 owner, tags, description이 있는가?
• 운영 DAG의 catchup 기본값이 의도와 맞는가?
• start_date가 동적으로 계산되지 않는가?
• task id가 팀 naming convention을 따르는가?
• 위험한 operator가 승인 없이 쓰이지 않았는가?
• 데이터셋을 쓰는 DAG가 lineage tag를 달고 있는가?
• SLA나 알림 정책이 빠지지 않았는가?

예를 들어 tag와 catchup 정책을 검사할 수 있다.

# tests/test_dag_policy.py
from airflow.models import DagBag

APPROVED_TAGS = {"sales", "billing", "platform", "quality", "adhoc"}


def all_dags():
    dagbag = DagBag(dag_folder="dags", include_examples=False)
    assert dagbag.import_errors == {}
    return dagbag.dags.values()


def test_dags_have_approved_tags():
    for dag in all_dags():
        assert dag.tags, f"{dag.dag_id} has no tags"
        unknown = set(dag.tags) - APPROVED_TAGS
        assert not unknown, f"{dag.dag_id} has unknown tags: {unknown}"


def test_production_dags_do_not_accidentally_catchup():
    for dag in all_dags():
        if "adhoc" not in dag.tags:
            assert dag.catchup is False, f"{dag.dag_id} catchup must be explicit"

구조 검사는 dependency wiring 실수를 잡는 데 유용하다.

def assert_downstream(dag, task_id, expected):
    task = dag.get_task(task_id)
    assert task.downstream_task_ids == set(expected)


def test_sales_dag_shape():
    dag = DagBag(dag_folder="dags", include_examples=False).get_dag("sales_daily")
    assert_downstream(dag, "extract", ["transform"])
    assert_downstream(dag, "transform", ["load", "quality_check"])
    assert_downstream(dag, "quality_check", ["publish"])

이 테스트의 목적은 Airflow 내부를 과하게 mock하는 것이 아니다. 운영자가 깨지면 안 된다고 판단한 DAG 구조와 정책을 코드로 기록하는 것이다.

4. dag.test(): 로컬에서 한 번의 실행 흐름을 빠르게 본다

Airflow 2.5 이후 계열에서 소개된 dag.test()는 DAG 파일에 작은 main block을 넣고, DAG를 단일 Python process에서 실행해 볼 수 있게 한다.

if __name__ == "__main__":
    dag.test()

공식 문서 기준으로 dag.test()는 executor를 띄우지 않고 task를 하나씩 실행한다. scheduler loop를 거치지 않기 때문에 빠르고, 실패하면 fail-fast에 가깝게 바로 확인할 수 있다. 로컬 SQLite 같은 테스트용 metadata DB에서도 사용할 수 있다.

실제 사용 감각은 다음과 같다.

export AIRFLOW_HOME=$(pwd)/.airflow-local
export AIRFLOW__CORE__LOAD_EXAMPLES=False
python dags/sales_daily.py

로컬 테스트의 목적은 production 전체를 흉내 내는 것이 아니다. 다음을 빠르게 확인하는 데 적합하다.

• Jinja template이 기대한 값으로 렌더링되는지
• task 간 XCom payload 형태가 맞는지
• 작은 fixture 데이터로 transformation이 성공하는지
• connection/variable mock이 빠지지 않았는지
• 특정 execution date 또는 data interval에서 분기 로직이 맞는지

반대로 다음은 dag.test()만으로 보장하기 어렵다.

• Celery/Kubernetes executor에서의 분산 실행 특성
• 실제 권한, network policy, service account 문제
• 대용량 데이터 성능
• scheduler backlog, pool, queue delay
• 외부 시스템의 rate limit과 장애 모드

따라서 dag.test()는 “운영 검증의 대체재”가 아니라 개발자 피드백 루프를 줄이는 도구로 봐야 한다.

5. CI 설계: 빠른 검사는 PR마다, 느린 검사는 승격 단계에서

Airflow DAG CI는 모든 것을 한 job에 넣으면 느려지고, 느리면 개발자가 우회하려 한다. 보통 두 층으로 나누는 편이 안정적이다.

5.1 PR gate: 빠르고 결정적인 검사

PR마다 돌리는 검사는 5~10분 안에 끝나는 것이 좋다.

# .github/workflows/airflow-dag-ci.yml
name: airflow-dag-ci

on:
  pull_request:
    paths:
      - "dags/**"
      - "plugins/**"
      - "requirements*.txt"
      - "tests/**"

jobs:
  dag-validation:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-test.txt
      - name: Validate DAGs
        env:
          AIRFLOW__CORE__LOAD_EXAMPLES: "False"
        run: |
          pytest tests/test_dag_imports.py tests/test_dag_policy.py

이 단계는 다음을 포함한다.

• Python format/lint/type check
• DagBag import error 검사
• DAG 구조와 팀 정책 검사
• custom operator/unit test
• fixture 기반 순수 transformation test

5.2 승격 gate: 실제 환경에 가까운 검사

main merge 뒤 또는 staging 배포 전에는 더 느린 검사를 둘 수 있다.

• Airflow image build
• provider package 충돌 확인
• staging metadata DB migration 확인
• airflow dags list / airflow tasks list <dag_id> smoke test
• canary DagRun 실행
• staging object storage나 test schema를 사용하는 end-to-end run

여기서 중요한 것은 production 데이터를 직접 건드리지 않는 것이다. DAG는 같은 코드라도 환경별 destination을 parameterize해야 한다. 공식 best practice도 staging 환경에서는 output path나 DB를 환경변수로 바꾸고, DAG 안에 값을 하드코딩하지 말라고 설명한다.

6. DAG 배포 방식과 rollback 전략

Airflow deployment에서 DAG 파일을 배포하는 방식은 환경마다 다르다. Apache Airflow Helm chart 문서는 대표적으로 세 가지 계열을 설명한다.

Helm chart 문서는 DAG를 image에 bake할 수 있지만, production에서 constant image tag를 쓰면 코드 히스토리를 잃기 때문에 좋지 않다고 경고한다. 운영 rollback을 생각하면 latest 같은 tag보다 commit SHA나 release tag가 낫다.

Airflow 3 계열에서는 DAG Bundle이 더 중요한 선택지가 된다. 공식 production deployment 문서는 multi-node 환경에서 Airflow가 DAG 파일 자체를 worker로 보내지 않고 “어떤 task를 실행하라”는 지시만 보내므로, 각 노드가 필요한 DAG와 설정을 갖고 있어야 한다고 설명한다. 그리고 DAG 동기화 방식으로 GitDagBundle을 포함한 DAG Bundle mechanism을 권장한다.

6.1 Airflow 3 DAG versioning과 rollback의 차이

Airflow 3의 DAG versioning은 UI와 실행 이력을 이해하는 데 큰 도움을 준다. 구조가 바뀐 DAG run은 버전과 연결되고, versioned DAG bundle을 쓰면 과거 run을 원래 bundle version으로 다시 실행할 수도 있다.

하지만 이것이 곧 “운영 rollback이 자동으로 끝난다”는 뜻은 아니다.

• DAG versioning: 어떤 DAG 구조와 코드가 어떤 run에 쓰였는지 추적한다.
• DAG bundle versioning: task가 특정 bundle version의 코드로 실행되도록 돕는다.
• 운영 rollback: 새 배포가 잘못됐을 때 scheduler, worker, dependency, connection, downstream side effect까지 포함해 안전한 이전 상태로 되돌리는 절차다.

예를 들어 새 DAG가 잘못된 테이블에 write했다면 Git commit을 되돌리는 것만으로 이미 쓴 데이터를 복구할 수 없다. rollback runbook에는 최소한 다음 질문이 있어야 한다.

1. 배포 artifact는 무엇인가? image tag, Git commit, bundle version, chart release 중 무엇을 되돌리는가?
2. 이미 시작된 DagRun은 중단할 것인가, 계속 둘 것인가?
3. 실패한 task는 clear할 것인가, 새 run으로 재처리할 것인가?
4. 외부 시스템에 이미 쓴 데이터는 idempotent하게 overwrite 가능한가?
5. schema 변경, connection 변경, variable 변경도 함께 되돌려야 하는가?

7. 배포 후 확인: import error 0에서 끝내지 않는다

배포가 끝난 뒤에는 최소한 세 가지를 확인한다.

7.1 Airflow가 새 DAG를 정상 parse했는가?

airflow dags list
airflow dags show sales_daily --save /tmp/sales_daily.png

또는 UI에서 import error가 없는지 확인한다. import error가 0이어도 DAG parse 시간이 급증했는지, scheduler heartbeat가 밀리는지 봐야 한다.

7.2 canary run이 예상 범위에서 끝나는가?

운영 DAG 전체를 바로 열기 전에 작은 입력 범위나 staging destination으로 canary를 돌린다.

airflow dags trigger sales_daily \
  --conf '{"mode":"canary","target_date":"2026-07-03"}'

canary run은 단순 성공/실패보다 다음을 봐야 한다.

• 실행 시간이 기존 범위 안인가?
• task retry가 비정상적으로 늘지 않았는가?
• destination row count나 partition count가 예상 범위인가?
• downstream dashboard나 quality check가 깨지지 않았는가?

7.3 관측성 대시보드에 배포 marker가 있는가?

DAG 배포는 scheduler 성능과 worker error rate에 영향을 준다. 배포 시각, Git SHA, image tag, bundle version을 대시보드 annotation으로 남기면 사고 대응 때 “언제부터 느려졌는가”를 빠르게 좁힐 수 있다.

8. 운영 체크리스트

PR 전 개발자 체크

• [ ] DAG 파일을 scheduler image와 같은 Python 버전·의존성에서 import했다.
• [ ] top-level network call, DB query, heavy import가 없다.
• [ ] dag.test() 또는 task unit test로 주요 분기 하나를 실행했다.
• [ ] fixture 데이터가 production destination으로 write하지 않는다.
• [ ] 새 connection, variable, secret이 필요한 경우 별도 migration/배포 순서가 적혀 있다.

CI gate 체크

• [ ] DagBag.import_errors == {} 검사가 있다.
• [ ] DAG 구조와 팀 정책 검사가 있다.
• [ ] requirements 변경이 Airflow image와 함께 검증된다.
• [ ] test가 Airflow 예제 DAG를 로드하지 않는다.
• [ ] flaky 외부 API 호출이 PR test에 직접 포함되지 않는다.

운영 배포 체크

• [ ] 배포 artifact가 immutable하다. 예: Git SHA, image tag, bundle version.
• [ ] rollback 대상과 명령이 runbook에 적혀 있다.
• [ ] 배포 후 import error, scheduler heartbeat, parse duration을 확인한다.
• [ ] canary run 또는 제한된 data interval 검증을 수행한다.
• [ ] 실패 시 clear/retry/revert/reprocess 기준이 구분되어 있다.

마무리: 테스트는 “Airflow를 믿지 못해서”가 아니라 변경 비용을 낮추기 위해 한다

DAG 테스트와 CI를 과하게 만들면 배포가 느려진다. 반대로 아무 검사 없이 배포하면 scheduler import error나 잘못된 task wiring을 운영에서 발견한다. 균형점은 명확하다.

• import와 정책 검사는 PR마다 빠르게 돌린다.
• 실제 실행 검사는 작은 fixture와 staging에서 제한적으로 돌린다.
• 배포 artifact는 되돌릴 수 있게 고정한다.
• rollback은 Git revert가 아니라 데이터 side effect까지 포함한 운영 절차로 쓴다.

Airflow는 DAG를 Python 코드로 다루기 때문에 자유도가 높다. 그 자유도는 테스트와 배포 discipline이 있을 때 장점이 되고, 없을 때 scheduler 장애와 데이터 오염으로 돌아온다.

References

• Apache Airflow Documentation, “Best Practices” — https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html
• Apache Airflow Documentation, “Testing DAGs with dag.test()” — https://airflow.apache.org/docs/apache-airflow/2.6.0/core-concepts/executor/debug.html
• Apache Airflow Documentation, “Production Deployment” — https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/production-deployment.html
• Apache Airflow Documentation, “Dag Bundles” — https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/dag-bundles.html
• Apache Airflow Helm Chart Documentation, “Manage DAG files” — https://airflow.apache.org/docs/helm-chart/stable/manage-dag-files.html
• Astronomer Docs, “Test Airflow DAGs” — https://www.astronomer.io/docs/learn/testing-airflow
• Astronomer Docs, “DAG writing best practices in Apache Airflow” — https://www.astronomer.io/docs/learn/dag-best-practices
• Astronomer Docs, “DAG Versioning and DAG Bundles” — https://www.astronomer.io/docs/learn/airflow-dag-versioning