DAG 번들

DAG 번들 (Dag Bundles)

이 페이지는 하나 이상의 DAG와 그 관련 파일을 묶는 컬렉션인 DAG 번들(Dag Bundle)을 다뤄요. DAG 번들은 로컬 디렉터리, Git 저장소, S3·GCS 같은 외부 시스템에서 DAG를 가져올 수 있으며 버전 관리를 지원해요. LocalDagBundle, GitDagBundle, S3DagBundle, GCSDagBundle 타입과 설정 방법, 커스텀 번들 작성법을 설명해요.

출처: 문서

본문

DAG 번들은 하나 이상의 DAG, 파일과 그 관련 파일(다른 Python 스크립트, 설정 파일 또는 다른 리소스)의 컬렉션이에요. DAG 번들은 로컬 디렉터리, Git 저장소 또는 다른 외부 시스템 같은 다양한 위치에서 DAG를 소싱할 수 있어요. 배포 관리자는 커스텀 소스를 지원하도록 자신만의 DAG 번들 클래스를 작성할 수도 있어요. 또한 Airflow 배포에 둘 이상의 DAG 번들을 정의할 수 있어 DAG를 더 잘 구성할 수 있어요. 번들을 더 높은 수준으로 유지하면 DAG가 실행되는 데 필요한 모든 것을 버전 관리할 수 있어요.

이는 Airflow 2 이하의 Dags folder와 비슷하지만 더 강력해요. Dags folder에서는 DAG가 로컬 디스크의 한 곳에 있어야 했고, DAG를 그곳에 가져오는 것은 전적으로 배포 관리자의 책임이었어요.

DAG 번들은 버전 관리를 지원하므로 Airflow가 특정 버전의 DAG 번들로 Task를 실행할 수 있게 해줘요. 이를 통해 run 중간에 DAG가 업데이트되어도 Dag run이 전체 run 동안 같은 코드를 사용할 수 있어요.

DAG 번들이 왜 중요할까요?

버전 관리 (Version Control): 버전 관리를 지원함으로써 DAG 번들은 Dag run이 run 중간에 DAG가 업데이트되어도 전체 run 동안 같은 코드를 사용할 수 있게 해요.
확장성 (Scalability): DAG 번들로 Airflow는 많은 수의 DAG를 논리적 단위로 조직화해 효율적으로 관리할 수 있어요.
유연성 (Flexibility): DAG 번들은 Git 저장소 같은 외부 시스템과의 원활한 통합을 가능하게 해 DAG를 소싱할 수 있어요.

DAG 번들의 종류

Airflow는 각각 특정 사용 사례에 맞는 여러 종류의 DAG 번들을 지원해요:

airflow.dag_processing.bundles.local.LocalDagBundle : 이 번들은 Dag 파일을 포함한 로컬 디렉터리를 참조해요. 개발·테스트 환경에 이상적이지만 번들의 버전 관리를 지원하지 않으므로, Task는 항상 최신 코드로 실행돼요.

airflow.providers.git.bundles.git.GitDagBundle : 이 번들은 Git 저장소와 통합해 Airflow가 저장소에서 직접 DAG를 가져올 수 있게 해요. GitDagBundle은 버전 관리를 지원해요.

airflow.providers.amazon.aws.bundles.s3.S3DagBundle : 이 번들은 Dag 파일을 포함한 S3 버킷을 참조해요. 번들의 버전 관리를 지원하지 않으므로, Task는 항상 최신 코드로 실행돼요.

airflow.providers.google.cloud.bundles.gcs.GCSDagBundle : 이 번들은 Dag 파일을 포함한 GCS 버킷을 참조해요. 번들의 버전 관리를 지원하지 않으므로, Task는 항상 최신 코드로 실행돼요.

DAG 번들 구성하기

DAG 번들은 dag_bundle_config_list에서 구성해요. 여기에 하나 이상의 DAG 번들을 추가할 수 있어요.

Warning

자격 증명은 Connection으로 참조하세요 — 인라인으로 넣지 마세요.

번들 kwargs는 [dag_processor] dag_bundle_config_list 설정에 저장되며, expose_config가 활성화되면 Airflow가 Config API를 통해 노출해요. 설정을 읽을 권한이 있는 사용자는 이 값을 있는 그대로 읽을 수 있으므로, 비밀(secret)을 포함하면 안 돼요.

자격 증명을 번들 kwargs에 직접 내장하지 마세요 — 예를 들어 https://x-access-token:<token...it처럼 repo_url에 직접 토큰을 넣는 방식요. 대신 Airflow Connection을 참조하고(Git용 git_conn_id, S3용 aws_conn_id, GCS용 gcp_conn_id) 자격 증명을 secrets backend에 보관해요. Connection 필드는 런타임에 해석되며 dag_bundle_config_list에 기록되지 않아요.

기본적으로 Airflow는 구성된 Dags 폴더를 가리키는 LocalDagBundle을 추가해 Airflow 2의 Dags folder와 같은 동작을 유지해요. 유일한 kwarg는 path이며, 생략하면 dags_folder 값이 기본값이 돼요:

[dag_processor]
dag_bundle_config_list = [
    {
      "name": "dags-folder",
      "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
      "kwargs": {
        "path": "/opt/airflow/dags"
      }
    }
  ]

Note

LocalDagBundle은 버전 관리를 지원하지 않아요. Task는 항상 디스크의 최신 코드로 실행돼요.

Git DAG 번들의 경우 필수 kwarg는 tracking_ref(브랜치, 태그 또는 커밋 SHA)예요. 저장소 자격 증명을 담은 Airflow connection을 참조하려면 git_conn_id를 사용하거나 repo_url을 직접 제공해요. subdir로 체크아웃을 하위 디렉터리로 좁힐 수도 있고, sparse_dirs로 특정 디렉터리의 sparse 체크아웃을 활성화할 수도 있어요:

[dag_processor]
dag_bundle_config_list = [
    {
      "name": "my-git-repo",
      "classpath": "airflow.providers.git.bundles.git.GitDagBundle",
      "kwargs": {
        "git_conn_id": "my_git_conn",
        "subdir": "dags",
        "tracking_ref": "main",
      }
    }
  ]

Note

GitDagBundle은 버전 관리를 지원해요. 각 Dag run은 생성된 Git 커밋을 기록하므로, 저장소가 이후 업데이트되어도 재실행이 정확히 같은 코드를 사용할 수 있어요.

전체 kwargs 목록과 더 많은 예제는 Bundles를 참고해요.

S3 DAG 번들의 경우 필수 kwarg는 bucket_name이에요. 선택적으로 aws_conn_id(기본값 aws_default)와 버킷 안의 하위 디렉터리로 번들을 한정하는 prefix를 설정할 수 있어요:

[dag_processor]
dag_bundle_config_list = [
    {
      "name": "my-s3-dags",
      "classpath": "airflow.providers.amazon.aws.bundles.s3.S3DagBundle",
      "kwargs": {
        "aws_conn_id": "aws_default",
        "bucket_name": "my-airflow-bucket",
        "prefix": "dags/"
      }
    }
  ]

Note

S3DagBundle은 버전 관리를 지원하지 않아요. Task는 항상 버킷의 최신 코드로 실행돼요.

전체 kwargs 목록과 더 많은 예제는 Bundles를 참고해요.

GCS DAG 번들의 경우 필수 kwarg는 bucket_name이에요. 선택적으로 gcp_conn_id(기본값 google_cloud_default)와 버킷 안의 하위 디렉터리로 번들을 한정하는 prefix를 설정할 수 있어요:

[dag_processor]
dag_bundle_config_list = [
    {
      "name": "my-gcs-dags",
      "classpath": "airflow.providers.google.cloud.bundles.gcs.GCSDagBundle",
      "kwargs": {
        "gcp_conn_id": "google_cloud_default",
        "bucket_name": "my-airflow-bucket",
        "prefix": "dags/"
      }
    }
  ]

Note

GCSDagBundle은 버전 관리를 지원하지 않아요. Task는 항상 버킷의 최신 코드로 실행돼요.

전체 kwargs 목록과 더 많은 예제는 Bundles를 참고해요.

하나의 배포에서 여러 번들 타입을 결합할 수 있어요. 기본 LocalDagBundle은 더 이상 필요 없으면 제거하거나, 다른 번들과 함께 유지할 수 있어요:

[dag_processor]
dag_bundle_config_list = [
    {
      "name": "my_git_bundle",
      "classpath": "airflow.providers.git.bundles.git.GitDagBundle",
      "kwargs": {"tracking_ref": "main", "git_conn_id": "my_git_conn"}
    },
    {
      "name": "dags-folder",
      "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
      "kwargs": {}
    }
  ]

Note

특히 마지막 줄의 공백은 멀티라인 값이 올바르게 작동하도록 중요해요. 더 자세한 내용은 configparser 문서에서 찾을 수 있어요.

Dag 번들이 제공하는 기본 view url과 다른 view url을 원한다면, Dag 번들 구성의 kwargs에서 url을 바꿀 수 있어요. 예를 들어 git Dag 번들에 커스텀 URL을 사용하려면:

[dag_processor]
dag_bundle_config_list = [
    {
      "name": "my_git_repo",
      "classpath": "airflow.providers.git.bundles.git.GitDagBundle",
      "kwargs": {
        "tracking_ref": "main",
        "git_conn_id": "my_git_conn",
        "view_url_template": "https://my.custom.git.repo/view/{subdir}",
      }
    }
  ]

위에서 view_url_template이 my_git_repo 번들의 DAG를 볼 때 사용될 커스텀 URL로 설정됐어요. {subdir} 플레이스홀더는 번들의 subdir 속성으로 대체돼요. 플레이스홀더는 번들의 속성들이에요. 번들의 속성 밖에 있는 플레이스홀더는 사용할 수 없어요. 커스텀 URL을 지정하면 Dag 번들이 제공하는 기본 URL을 덮어써요.

url은 안전성 검증을 거치며, 안전하지 않으면 번들의 view url은 None으로 설정돼요. 이는 안전하지 않은 URL로 인한 잠재적 보안 문제를 방지하기 위해서예요.

또한 kwargs로 전달해 refresh_interval을 Dag 번들별로 오버라이드할 수 있어요. 이는 Dag processor가 Dag 번들에서 새 파일을 새로고침하거나 찾는 빈도를 제어해요.

Airflow 3.0.2부터 git은 기본 이미지에 미리 설치돼요. 하지만 3.0.2 이전 버전을 사용한다면 docker 이미지에 git을 설치해야 해요.

RUN apt-get update && apt-get install -y git
ENV GIT_PYTHON_GIT_EXECUTABLE=/usr/bin/git
ENV GIT_PYTHON_REFRESH=quiet

사용자 가장(impersonation)과 함께 DAG 번들 사용하기

DAG 번들과 run_as_user(사용자 가장)를 함께 사용할 때는, 가장된 사용자가 메인 Airflow 프로세스가 만든 번들 파일에 접근할 수 있도록 적절한 파일 권한을 구성해야 해요.

모든 가장된 사용자와 Airflow 사용자는 같은 그룹이어야 해요.
적절한 umask 설정을 구성해요 (예: umask 0002).

Note

이 권한 기반 접근 방식은 임시 해결책이에요. 향후 Airflow 버전은 supervisor 기반 번들 작업을 통해 멀티 사용자 접근을 처리할 것이며, 공유 그룹 권한이 필요 없어질 거예요.

기본 재실행 버전 동작 구성

사용자가 DAG run이나 task instance를 clear할 때, UI는 최신 번들 버전으로 재실행할지 아니면 원래 run이 사용한 버전으로 재실행할지 묻는 체크박스를 표시해요. rerun_with_latest_version 설정은 그 체크박스의 기본 상태를 제어해서, 팀이 매번 수동으로 결정할 필요가 없게 해요. 이 같은 설정은 API나 CLI로 backfill을 만들 때 기본 run_on_latest_version 동작도 좌우해요.

Note

이는 버전 관리되는 번들 타입(GitDagBundle 같은)에만 적용돼요. 로컬 번들(LocalDagBundle)은 버전 관리를 지원하지 않으며 항상 최신 코드를 사용해요.

작동 방식

각 DAG는 dag processor가 DAG 파일을 재파싱할 때마다 업데이트되는 파싱 버전(DagModel.bundle_version)을 가져요. 각 DAG run은 생성된 번들 버전을 기록해요.

rerun_with_latest_version이 False일 때 DAG run을 clear하면 원래 번들 버전이 보존되어 재실행이 같은 코드를 사용해요. 이는 실패 디버깅 시 재현성을 제공해요. True일 때 clear하면 DAG run이 현재 파싱 버전으로 업데이트되어 재실행 시 가장 최근 코드가 사용되게 해요.

이 설정은 다음 우선순위(높은 것에서 낮은 것)로 해석돼요:

명시적 요청: API 요청 본문의 run_on_latest_version 파라미터 (제공된 경우)
DAG 레벨: DAG의 rerun_with_latest_version 파라미터 (True 또는 False인 경우)
글로벌 설정: [core] rerun_with_latest_version 옵션 (설정된 경우)
호출 지점별 폴백: clear/rerun은 False, backfill은 True (각 경로의 역사적 기본값 보존)

한 가지 예외: 자체 버전이 없는 Dag run — Airflow 2에서 이어받았거나 airflow db clean이 그 버전을 제거한 경우 — 은 보존할 것이 없으므로, 해석된 설정과 무관하게 항상 최신 버전과 번들 버전을 사용해요.

글로벌 설정

[core] rerun_with_latest_version 옵션으로 조직 전역 기본값을 설정해요:

[core]
rerun_with_latest_version = False  # Rerun with the original bundle version
# rerun_with_latest_version = True  # Rerun with the latest bundle version

설정하지 않으면 호출 지점이 역사적 폴백(clear/rerun은 False, backfill은 True)을 적용해요.

DAG 레벨 설정

특정 DAG의 글로벌 기본값을 오버라이드해요:

from datetime import datetime

from airflow import DAG
from airflow.operators.empty import EmptyOperator

# Always rerun with the latest version
with DAG(
    dag_id="always_latest_dag",
    rerun_with_latest_version=True,
    start_date=datetime(2024, 1, 1),
) as dag:
    EmptyOperator(task_id="task")

사용 사례

실패한 run 디버깅하기: : False(기본값)로 설정하면 실패한 run을 clear할 때 같은 코드로 재실행해 문제를 재현·격리하기 쉬워요.

항상 최신 코드 실행하기: : 팀이 재실행이 항상 최신 코드를 사용하길 선호한다면 [core] rerun_with_latest_version = True로 설정해요 — 예를 들어 원래 run 이후 버그 수정이 배포된 경우요.

혼합 정책: : 글로벌 기본값을 True로 설정하되, 버전 안정성이 중요한 특정 핵심 DAG는 rerun_with_latest_version=False로 오버라이드해요.

`disable_bundle_versioning`과의 관계

Airflow는 번들 버전 관리 동작에 영향을 주는 두 개의 별도 설정을 제공해요. 그것들은 서로 다른 목적을 가져요:

disable_bundle_versioning : 버전 추적을 완전히 끄는 설정. True로 설정하면 DAG run에 bundle_version이 기록되지 않아요. DAG 파라미터와 글로벌 설정 옵션([dag_processor] disable_bundle_versioning)으로 사용할 수 있어요.

rerun_with_latest_version : 버전 추적을 활성화로 유지하면서 기본 재실행 동작을 제어해요. 사용자가 Task를 clear하거나 재실행할 때 새 run이 최신 번들 버전을 사용할지 원래 버전을 사용할지 결정해요. 버전 관리는 활성화된 채 유지되므로 버전 기록은 여전히 기록돼요. 이것은 단지 사용자에게 제시되는 기본 선택만 바꿔요.

간단히 말하면: disable_bundle_versioning은 "아예 버전을 추적할까?"에 답하고, rerun_with_latest_version은 "재실행할 때 어떤 버전이 기본일까?"에 답해요. 두 설정은 독립적이며, rerun_with_latest_version은 버전 관리가 비활성화되면 효과가 없어요.

커스텀 DAG 번들 작성하기

BaseDagBundle 클래스를 확장해 나만의 DAG 번들을 구현할 때 구현해야 할 몇 가지 메서드가 있어요. 아래는 커스텀 DAG 번들을 구현하는 데 도움을 주는 가이드예요.

추상 메서드

다음 메서드는 추상이며 커스텀 번들 클래스에서 구현해야 해요:

path : 이 속성은 이 번들에 대한 Dag 파일이 저장된 디렉터리의 Path를 반환해야 해요. Airflow는 이 속성을 사용해 처리할 Dag 파일을 찾아요.

get_current_version : 이 메서드는 번들의 현재 버전을 문자열로 반환해야 해요. Airflow는 나중에 Task 실행 시 이 버전을 다시 얻기 위해 __init__에 이 버전을 전달해요. 버전 관리를 지원하지 않으면 None을 반환해야 해요.

refresh : 이 메서드는 소스에서 번들의 내용을 새로고침하는 작업(예: 원격 저장소에서 최신 변경 사항 가져오기)을 처리해야 해요. Dag processor가 번들이 최신 상태인지 확인하기 위해 주기적으로 사용해요.

선택적 메서드

추상 메서드에 더해, 번들 동작을 커스터마이즈하기 위해 다음 메서드를 오버라이드할 수도 있어요:

init : 이 메서드는 GitDagBundle의 tracking_ref 같은 추가 파라미터로 번들을 초기화하도록 확장할 수 있어요. 또한 올바른 초기화를 위해 부모 클래스의 __init__ 메서드도 호출해야 해요. 네트워크 호출 같은 비싼 작업은 번들 인스턴스화 중 지연을 방지하기 위해 이 메서드에서 피해야 하며, initialize 메서드에서 해야 해요.

initialize : 이 메서드는 번들이 Dag processor나 worker에서 처음 사용되기 전에 호출돼요. 번들의 내용이 접근될 때만 비싼 작업을 수행할 수 있게 해요.

view_url : 이 메서드는 외부 시스템(예: Git 저장소의 웹 인터페이스)에서 번들을 볼 URL을 문자열로 반환해야 해요.

기타 고려 사항

버전 관리: 번들이 버전 관리를 지원한다면, initialize, get_current_version, refresh가 버전별 로직을 처리하도록 구현했는지 확인해요.
동시성: Worker는 동시에 많은 번들을 만들 수 있으며, 번들 객체에 대한 호출을 직렬화하지 않아요. 따라서 기반 기술에서 문제가 된다면 번들 클래스가 잠금(locking)을 처리해야 해요. 예를 들어 git 저장소를 클로닝한다면, 한 번에 하나의 번들 객체만 클로닝하도록 번들 클래스가 잠금을 책임져야 해요. 필요하다면 이 목적으로 사용할 수 있는 기본 클래스의 lock 메서드가 있어요.
Triggerer 제한: DAG 번들은 triggerer 컴포넌트에서 초기화되지 않아요. 실질적으로는 trigger가 DAG 번들에서 올 수 없다는 뜻이에요. triggerer는 모든 것이 메인 프로세스에서 일어나므로 시간이 지나면서 trigger 코드 변경을 다루지 않기 때문이에요. 대신 trigger는 sys.path의 다른 어디에서든 올 수 있어요. 커스텀 trigger를 사용해야 한다면 DAG 번들에서 소싱하지 말고 Python 환경의 sys.path에서 사용할 수 있는지 확인해요.

DAG 번들

DAG 번들 (Dag Bundles)

본문

DAG 번들이 왜 중요할까요?

DAG 번들의 종류

DAG 번들 구성하기

사용자 가장(impersonation)과 함께 DAG 번들 사용하기

기본 재실행 버전 동작 구성

작동 방식

글로벌 설정

DAG 레벨 설정

사용 사례

disable_bundle_versioning과의 관계

커스텀 DAG 번들 작성하기

추상 메서드

선택적 메서드

기타 고려 사항

더 알아보기 (Learn more)

`disable_bundle_versioning`과의 관계