[Airflow] DAG 처음 짜면서 했던 실수들

1. 들어가며

Airflow를 처음 접했을 때 솔직히 이런 생각이 들었다.

"이거 그냥 cron이랑 뭐가 달라?"

Python으로 스케줄 짜고, 태스크 실행하고… 겉으로 보면 비슷해 보인다. 그런데 실제로 DAG를 짜기 시작하자마자 예상치 못한 곳에서 계속 막혔다.

이 글은 그 과정에서 내가 직접 겪었던 실수들을 정리한 기록이다. Airflow를 처음 시작하는 분들이 같은 곳에서 시간을 낭비하지 않았으면 하는 마음으로 썼다.

---

2. start_date를 datetime.now()로 설정했다

어떤 실수였나

DAG를 처음 만들 때 start_date를 이렇게 설정했다.

 

 

python

from datetime import datetime
from airflow import DAG

with DAG(
    dag_id="my_first_dag",
    start_date=datetime.now(),  # 이렇게 하면 안 된다
    schedule_interval="@daily",
) as dag:
    ...

언뜻 보면 "지금부터 시작"이라는 뜻이니까 자연스러워 보인다. 그런데 이걸 실행하면 DAG가 제대로 트리거되지 않거나, 매번 실행할 때마다 start_date가 달라지는 문제가 생긴다. 그렇게 되면 과거 시점이 존재하지 않게되어 catchup이 불가능해지는 문제도 있다.

왜 안 되는 걸까

Airflow는 DAG 파일을 주기적으로 파싱한다. 스케줄러가 DAG 파일을 읽을 때마다 datetime.now()가 새로 계산되기 때문에, start_date가 계속 바뀐다. Airflow 입장에서는 "언제부터 시작해야 하는지"를 일관되게 알 수 없게 되는 것이다.

 

올바른 방법

start_date는 항상 고정된 과거 날짜로 설정한다.

 

 

python

from datetime import datetime
from airflow import DAG

with DAG(
    dag_id="my_first_dag",
    start_date=datetime(2024, 1, 1),  # 고정된 과거 날짜
    schedule_interval="@daily",
) as dag:
    ...

---

3. catchup=True인 줄 몰랐다

어떤 실수였나

DAG를 처음 활성화했더니 갑자기 태스크가 수십 개씩 한꺼번에 실행되기 시작했다. "왜 이렇게 많이 돌아가지?" 하고 보니, Airflow가 start_date부터 현재까지 밀린 실행을 모두 메꾸려고 한 것이었다.

예를 들어 start_date를 6개월 전으로 설정하고 schedule_interval="@daily"로 하면, DAG를 활성화하는 순간 180개의 DAG Run이 한꺼번에 생성된다.

왜 이런 일이 생기나

Airflow에는 catchup이라는 설정이 있는데, 기본값이 True 다. catchup=True는 "과거에 실행됐어야 했는데 못 한 것들을 지금 다 채워라"라는 의미다.

처음엔 이 기본값의 존재 자체를 몰랐다.

올바른 방법

대부분의 경우, 특히 처음 개발할 때는 catchup=False로 설정하는 것이 안전하다.

 

 

python

with DAG(
    dag_id="my_first_dag",
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False,  # 과거 실행 채우지 않음
) as dag:
    ...

물론 catchup=True가 유용한 경우도 있다. 데이터 파이프라인에서 특정 날짜 범위의 데이터를 재처리해야 할 때는 의도적으로 True로 설정하기도 한다. 중요한 건 기본값이 True라는 것을 알고 명시적으로 선택하는 것이다.

---

4. Task가 실패해도 데이터가 중복됐다

어떤 실수였나

Task가 중간에 실패해서 재실행했더니 DB에 데이터가 두 번 들어가 있었다. "재실행하면 이전 결과가 덮어씌워지는 거 아닌가?" 하고 생각했는데, 전혀 그렇지 않았다.

 

 

python

def insert_data():
    # 단순 INSERT는 재실행 시 중복 발생
    db.execute("INSERT INTO orders SELECT * FROM raw_orders WHERE date = '2024-01-01'")

Task가 실패한 후 재실행하면, 이전에 이미 실행된 부분이 또 실행되면서 데이터가 쌓였다.

핵심 개념 - 멱등성(Idempotency)

이때 처음 알게 된 개념이 멱등성(Idempotency) 이다.

멱등성이란, 같은 작업을 몇 번 반복해도 결과가 동일한 성질을 말한다. Airflow에서 Task는 실패 시 재실행될 수 있기 때문에, Task 자체가 멱등해야 한다. 즉, "몇 번 실행해도 결과가 같아야 한다"는 원칙이다.

올바른 방법

단순 INSERT 대신 UPSERT나 삭제 후 재삽입 방식을 사용한다.

 

 

python

def insert_data():
    # 방법 1: UPSERT (PostgreSQL 기준)
    db.execute("""
        INSERT INTO orders SELECT * FROM raw_orders WHERE date = '2024-01-01'
        ON CONFLICT (id) DO UPDATE SET ...
    """)

def insert_data():
    # 방법 2: 해당 날짜 데이터 삭제 후 재삽입
    db.execute("DELETE FROM orders WHERE date = '2024-01-01'")
    db.execute("INSERT INTO orders SELECT * FROM raw_orders WHERE date = '2024-01-01'")

처음엔 "그냥 재실행 안 되게 잘 짜면 되지 않나?"라고 생각할 수 있다. 하지만 분산 환경에서 Task는 언제든 실패할 수 있고, Airflow는 그 상황에서도 파이프라인이 안전하게 재실행될 수 있도록 설계되어야 한다.

---

5. 마치며

지금 돌이켜보면 이 실수들은 다 "당연히 이렇게 되겠지"라는 막연한 가정에서 비롯됐다.

• datetime.now()가 직관적으로 보였고,
• catchup 같은 기본값의 존재를 몰랐고,
• 재실행 시 멱등성이 보장돼야 한다는 개념 자체가 낯설었다.

Airflow는 단순히 스크립트를 스케줄링하는 도구가 아니라, 분산 환경에서 신뢰할 수 있는 데이터 파이프라인을 만들기 위한 도구다.

그 관점에서 다시 보면 위의 개념들이 왜 그렇게 설계됐는지 자연스럽게 이해된다.

 

다음엔 Airflow를 조금 더 써보면서 느낀 것들을 정리해볼 예정이다.