Airflow 베스트 프랙티스 — 안정적인 DAG를 만드는 요령

Airflow 베스트 프랙티스 — 안정적인 DAG를 만드는 요령

DAG 하나 만드는 것 자체는 쉬워요. 하지만 그 DAG가 재실행되거나 실패했을 때 예상 밖의 결과를 내지 않으려면, 처음부터 지킬 규칙들이 있어요. 이 문서는 DAG 작성부터 환경 의존성 구성까지, 운영에서 빈번히 마주치는 함정을 피하는 관행을 소개해요.

출처: Best Practices — Airflow Documentation

본문

DAG 만들기의 세 단계

새 DAG를 만드는 건 크게 세 단계로 나뉘어요.

• DAG 객체를 만드는 Python 코드를 작성하기
• 코드가 기대대로 동작하는지 테스트하기
• DAG를 실행할 환경 의존성을 설정하기

태스크 만들기: 데이터베이스 트랜잭션처럼

Airflow의 태스크는 데이터베이스의 트랜잭션과 비슷하게 다뤄야 해요. 태스크가 불완전한 결과를 만들어내는 일은 없어야 해요. 예를 들어 태스크가 끝났는데 HDFS나 S3에 불완전한 데이터가 남아 있으면 안 되죠.

Airflow는 태스크가 실패하면 재시도할 수 있어요. 그래서 태스크는 매번 다시 실행해도 같은 결과를 내도록 해야 해요. 결과가 달라지는 것을 피하는 방법은 몇 가지가 있어요.

• 재실행 시 INSERT를 쓰지 마세요. INSERT는 데이터베이스에 중복 행을 만들 수 있어요. UPSERT로 바꿔주세요.
• 특정 파티션에서 읽고 쓰세요. 태스크에서 "가장 최신 데이터"를 무조건 읽지 마세요. 재실행 사이에 입력 데이터가 바뀌면 결과가 달라질 수 있거든요. 더 나은 방법은 특정 파티션의 입력 데이터를 읽는 것이고, data_interval_start를 파티션으로 쓸 수 있어요. S3/HDFS에 쓸 때도 이 파티셔닝 방식을 지키세요.
• Python의 now()는 현재 시각 객체를 주는데, 태스크 안에서 특히 중요한 계산에 쓰면 안 돼요. 실행할 때마다 결과가 달라지기 때문이에요. 임시 로그를 만들 때 같은 용도로 쓰는 건 괜찮아요.

반복되는 파라미터(예: connection_id나 S3 경로)는 태스크마다 새로 선언하기보다 default_args에 정의하는 게 좋아요. default_args는 오타 같은 실수를 막아줘요. 또 대부분의 커넥션 타입은 태스크에서 파라미터 이름이 고유하니(예: gcp_conn_id), 커넥션을 default_args에 한 번만 선언하면 그 커넥션 타입을 쓰는 모든 오퍼레이터가 자동으로 사용해요.

태스크 삭제할 때

DAG에서 태스크를 삭제할 때는 조심해야 해요. Graph View나 Grid View 등에서 태스크가 보이지 않게 돼서, Webserver에서 그 태스크의 로그를 확인하기 어려워져요. 그러한 기록이 꼭 필요하다면, 새 DAG를 만들어 사용하는 편이 나아요.

태스크 간 통신

Kubernetes executor나 Celery executor를 쓰면 DAG의 태스크가 서로 다른 서버에서 실행돼요. 따라서 로컬 파일시스템에 파일이나 설정을 저장하면 안 돼요. 하위 태스크가 다른 서버에서 돌아서 그 파일에 접근할 수 없을 수 있기 때문이에요. LocalExecutor여도 디스크에 파일을 저장하면 재시도가 어려워질 수 있어요(예: 다른 태스크가 설정 파일을 지워버리는 경우).

가능하면 작은 메시지는 XCom으로 주고받고, 큰 데이터는 S3/HDFS 같은 원격 스토리지를 쓰는 게 좋아요. 예를 들어 처리된 데이터를 S3에 저장하는 태스크는 결과 데이터의 S3 경로를 Xcom으로 밀어 넣고, 하위 태스크는 그 경로를 꺼내서 읽는 방식이에요.

태스크 안에 비밀번호나 토큰 같은 인증 파라미터를 저장해서도 안 돼요. 가능하면 Connections를 써서 Airflow 백엔드에 안전하게 보관하고, 고유 커넥션 id로 꺼내 쓰세요.

최상위 레벨(top-level) Python 코드 줄이기

DAG를 만들 때는 오퍼레이터를 만들고 DAG 관계를 구성하는 데 필요한 코드 외에는 최상위 레벨 코드를 두지 않는 게 좋아요. 스케줄러의 설계상, 최상위 코드의 파싱 속도가 Airflow의 성능과 확장성에 영향을 주기 때문이에요.

Airflow 스케줄러는 오퍼레이터의 execute 메서드 밖의 코드를 min_file_process_interval초 간격으로 실행해요. 이는 동적 스케줄링을 가능하게 하기 위한 것인데, 스케줄·의존성이 시간에 따라 바뀌어 DAG의 다음 스케줄에 영향을 줄 수 있기 때문이에요. 그래서 스케줄러가 DAG의 내용이 스케줄된 태스크에 올바르게 반영되도록 계속 확인해요.

구체적으로는 데이터베이스 접근, 무거운 계산, 네트워킹 연산을 최상위에서 돌리면 안 돼요.

Python 개발자가 흔히 놓치는 것 하나는, 최상위 레벨의 import가 예상보다 오래 걸릴 수 있고 오버헤드를 만든다는 점이에요. 이는 callable 안에서 로컬 import로 바꾸면 쉽게 피할 수 있어요.

최상위 DAG 코드를 피하지 않은 예:

import pendulum

from airflow.sdk import DAG
from airflow.sdk import task


def expensive_api_call():
    print("Hello from Airflow!")
    sleep(1000)


my_expensive_response = expensive_api_call()

with DAG(
    dag_id="example_python_operator",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:

    @task()
    def print_expensive_api_call():
        print(my_expensive_response)

최상위 DAG 코드를 피한 예:

import pendulum

from airflow.sdk import DAG
from airflow.sdk import task


def expensive_api_call():
    sleep(1000)
    return "Hello from Airflow!"


with DAG(
    dag_id="example_python_operator",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:

    @task()
    def print_expensive_api_call():
        my_expensive_response = expensive_api_call()
        print(my_expensive_response)

첫 번째 예에서는 expensive_api_call이 DAG 파일이 파싱될 때마다 실행돼서 파일 처리 성능이 나빠져요. 두 번째 예에서는 태스크가 실제로 실행될 때만 호출되니 성능 영향 없이 파싱돼요.

import 문도 최상위 코드로 취급된다는 점을 기억하세요. 오래 걸리는 import나, import된 모듈 자체가 최상위에서 코드를 실행한다면 스케줄러 성능에 영향을 줄 수 있어요. 다음과 같이 처리할 수 있어요:

# DAG 파일 최상위에서 import해도 부담 없는 모듈
import random
import pendulum

# 무거운 import는 최상위에서 피하는 게 좋다. DAG 파일은 자주 파싱되므로 최상위 코드가 반복 실행된다.
#
# import pandas
# import torch
# import tensorflow
#


@task()
def do_stuff_with_pandas_and_torch():
    import pandas
    import torch

    # do some operations using pandas and torch


@task()
def do_stuff_with_tensorflow():
    import tensorflow

    # do some operations using tensorflow

내 코드가 "최상위"인지 확인하는 법

코드가 최상위인지를 이해하려면 파이썬 파싱이 어떻게 동작하는지 알아야 해요. 일반적으로 Python이 파일을 파싱할 때, 메서드 내부 코드는 실행하지 않고 그 외의 보이는 코드를 실행해요. 메서드의 기본값을 정하는 데 쓰이는 코드도 최상위에 해당하는 등, 쉽게 알 수 없는 특수한 경우도 많아요.

쉬운 확인 방법이 있어요. 파싱했을 때 그 코드 조각이 실제로 실행되는지를 보면 돼요. 아래 코드를 생각해볼게요:

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
import pendulum


def get_task_id():
    return "print_array_task"  # <- is that code going to be executed?


def get_array():
    return [1, 2, 3]  # <- is that code going to be executed?


with DAG(
    dag_id="example_python_operator",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:
    operator = PythonOperator(
        task_id=get_task_id(),
        python_callable=get_array,
        dag=dag,
    )

확인하려면 확인하고 싶은 코드에 print 문을 넣고 python <my_dag_file>.py를 실행해보면 돼요.

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
import pendulum


def get_task_id():
    print("Executing 1")
    return "print_array_task"  # <- is that code going to be executed? YES


def get_array():
    print("Executing 2")
    return [1, 2, 3]  # <- is that code going to be executed? NO


with DAG(
    dag_id="example_python_operator",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:
    operator = PythonOperator(
        task_id=get_task_id(),
        python_callable=get_array,
        dag=dag,
    )

실행해보면 이런 출력이 보여요:

[Breeze:3.10.19] root@cf85ab34571e:/opt/airflow# python /files/test_python.py
Executing 1

즉 get_array는 최상위 코드로 실행되지 않고, get_task_id는 실행되는 것을 알 수 있어요.

코드 품질과 Lint

높은 코드 품질 유지는 Airflow 워크플로의 신뢰성과 유지보수성에 중요해요. 요점을 정리하면:

1. 이 문서가 다루는 베스트 프랙티스 중 일부는 Ruff AIR 규칙으로도 지원돼요. 이 규칙들은 Airflow 특유의 베스트 프랙티스(비권장 패턴·마이그레이션 이슈 등)를 감지·강제해요. 전체 목록은 Airflow (AIR)에서 볼 수 있어요.
2. 새 베스트 프랙티스를 제안하고 대응하는 Ruff AIR 규칙을 추가하려면, Proposing Airflow Best Practices and Ruff AIR Rules의 기여 절차를 따라주세요.

ruff 설치와 사용

1. pip로 ruff를 설치해요:

   pip install "ruff>=0.15.17"
   

2. DAG에 잠재적 문제가 없는지 확인해요:

   ruff check dags/ --select AIR3
   

   이 명령은 dags/ 디렉터리의 DAG를 분석해 지정한 규칙과 관련된 이슈를 보고해요.

예시

레거시 DAG가 이렇게 정의되어 있다면:

from airflow import dag
from airflow.datasets import Dataset
from airflow.sensors.filesystem import FileSensor


@dag()
def legacy_dag():
    FileSensor(task_id="wait_for_file", filepath="/tmp/test_file")

ruff 실행 결과는 이렇게 나와요:

dags/legacy_dag.py:7:2: AIR301 Dag should have an explicit schedule argument
dags/legacy_dag.py:12:6: AIR302 schedule_interval is removed in Airflow 3.0
dags/legacy_dag.py:17:15: AIR302 airflow.datasets.Dataset is removed in Airflow 3.0
dags/legacy_dag.py:19:5: AIR303 airflow.sensors.filesystem.FileSensor is moved into ``standard`` provider in Airflow 3.0

ruff를 개발 워크플로에 통합하면 비권장 패턴을 미리 해결하고 코드 품질을 유지해서, Airflow 버전 간 전환을 더 매끄럽게 만들어요.

DAG 정적 타입 체크

mypy로 DAG 타입을 체크한다면, 선택적인 apache-airflow-mypy 플러그인이 타이핑 데코레이터나 오퍼레이터 출력 같은 Airflow 특유 패턴에 정확한 결과를 줘요. 자세한 설치·설정은 Static Type Checking for Dags를 보세요.

동적 DAG 생성

DAG를 손으로 일일이 쓰는 게 현실적이지 않을 때가 있어요. 파라미터 하나만 달라지는 비슷한 DAG가 많이 필요하거나, 테이블을 로드하는 DAG 묶음이 필요한데 테이블이 바뀔 때마다 DAG를 일일이 수정하고 싶지 않은 상황이 그렇죠. 이런 경우 DAG를 동적으로 생성하는 편이 유용해요.

앞서 설명한 최상위 코드 처리 부하를 줄이는 것은 동적 DAG 구성에서 특히 중요해요. 동적 구성은 크게 다음 중 하나로 만들 수 있어요.

• 환경 변수로 (Airflow Variables인 core-concepts/variables와 혼동하지 마세요)
• DAG 폴더 안에 메타데이터를 담은 외부 생성 Python 코드로
• DAG 폴더 안의 외부 생성 설정 메타데이터 파일로

동적 DAG 생성의 몇몇 사례는 Dynamic Dag Generation에서 볼 수 있어요.

Airflow Variables 사용법

Airflow Variables를 쓰면 네트워크 호출과 데이터베이스 접근이 발생해요. 그래서 앞서 말한 최상위 Python 코드에서의 사용은 최대한 피해야 해요. 최상위 DAG 코드에서 꼭 써야 한다면, 적절한 ttl로 설정된 실험적 캐시를 켜서 DAG 파싱 영향을 줄일 수 있어요.

오퍼레이터의 execute() 메서드 안에서는 Variables를 자유롭게 쓸 수 있고, Jinja 템플릿을 통해 기존 오퍼레이터에 넘겨 태스크 실행 시점까지 읽기를 미룰 수도 있어요.

템플릿 문법은 이렇게 써요:

{{ var.value.<variable_name> }}

변수에서 json 객체를 역직렬화해야 한다면:

{{ var.json.<variable_name> }}

최상위 코드에서 Jinja 템플릿을 쓰면 태스크가 실행되기 전까지 요청이 발생하지 않아요. 반면 Variable.get()은 캐시가 켜져 있지 않으면 스케줄러가 DAG 파일을 파싱할 때마다 요청을 만들어요. 캐시 없이 Variable.get()을 쓰면 DAG 파일 처리 성능이 떨어지고, 경우에 따라 파싱 완료 전에 DAG 파일이 타임아웃될 수도 있어요.

나쁜 예:

from airflow.sdk import Variable

foo_var = Variable.get("foo")  # AVOID THAT
bash_use_variable_bad_1 = BashOperator(
    task_id="bash_use_variable_bad_1", bash_command="echo variable foo=${foo_env}", env={"foo_env": foo_var}
)

bash_use_variable_bad_2 = BashOperator(
    task_id="bash_use_variable_bad_2",
    bash_command=f"echo variable foo=${Variable.get('foo')}",  # AVOID THAT
)

bash_use_variable_bad_3 = BashOperator(
    task_id="bash_use_variable_bad_3",
    bash_command="echo variable foo=${foo_env}",
    env={"foo_env": Variable.get("foo")},  # AVOID THAT
)

좋은 예:

bash_use_variable_good = BashOperator(
    task_id="bash_use_variable_good",
    bash_command="echo variable foo=${foo_env}",
    env={"foo_env": "{{ var.value.get('foo') }}"},
)

@task
def my_task():
    var = Variable.get("foo")  # This is ok since my_task is called only during task run, not during Dag scan.
    print(var)

보안 측면에서, 민감한 데이터를 담은 변수는 Secrets Backend를 쓰는 것을 권장해요.

Timetable 주의점

timetable 코드의 최상위 레벨에서 Airflow Variables/Connections를 쓰거나 Airflow 데이터베이스에 접근하는 걸 피하세요. 데이터베이스 접근은 DAG 실행 시점까지 미뤄야 해요. timetable 클래스 초기화의 인자로 변수/커넥션 조회를 넣거나, 커스텀 timetable 모듈 최상위에 Variable/커넥션을 두면 안 돼요.

나쁜 예:

from airflow.sdk import Variable
from airflow.timetables.interval import CronDataIntervalTimetable


class CustomTimetable(CronDataIntervalTimetable):
    def __init__(self, *args, something=Variable.get("something"), **kwargs):
        self._something = something
        super().__init__(*args, **kwargs)

좋은 예:

from airflow.sdk import Variable
from airflow.timetables.interval import CronDataIntervalTimetable


class CustomTimetable(CronDataIntervalTimetable):
    def __init__(self, *args, something="something", **kwargs):
        self._something = Variable.get(something)
        super().__init__(*args, **kwargs)

변경 후 DAG 트리거하기

DAG나 DAG 폴더의 다른 파일을 바꾼 직후에는 바로 트리거하지 않는 게 좋아요. 시스템이 변경된 파일을 처리할 충분한 시간을 줘야 해요. 먼저 파일이 스케줄러에 배포되고(보통 분산 파일시스템이나 Git-Sync), 스케줄러가 Python 파일을 파싱해 데이터베이스에 저장하는 여러 단계를 거쳐요. 구성에 따라 이 과정은 수 초에서 수 분, 극단적인 경우 수십 분까지 걸릴 수 있어요. DAG가 UI에 나타나는 걸 확인한 뒤에 트리거해야 해요.

Watcher 패턴과 트리거 규칙

Watcher 패턴은 다른 태스크들의 상태를 "지켜보는" 태스크가 있는 DAG를 부르는 말이에요. 주요 목적은 다른 태스크가 실패했을 때 DAG Run을 실패시키는 것이에요. Airflow 시스템 테스트에서 생겨난 아이디어로, 여러 태스크(테스트의 단계들처럼)를 가진 DAG에서 유용해요.

더 알아보기

• Best Practices (Airflow 공식 문서)
• Variables (변수) 위키 페이지
• 오퍼레이터 (Operators) 위키 페이지
• 태스크 (Tasks) 위키 페이지
• XCom 위키 페이지