[Airflow 책 3장] Airflow의 스케줄링: 매일 실행하면 하루치 데이터도 알아서 가져올까? (Airflow 3 반영)

vividbaek · 10월 30, 2026, 7:51오전

2장에서는 DAG를 만들고, 실행하고, 실패한 태스크를 다시 돌려 봤습니다. 그런데 날짜는 여전히 헷갈렸습니다. start_date를 적으면 그날 바로 실행되는 건지, 오늘 도는 DAG에는 왜 어제 날짜가 붙는지, 잠깐 꺼 두었다가 켜면 그동안의 데이터도 처리하는지 궁금했습니다. 《Apache Airflow 기반의 데이터 파이프라인》 3장은 사용자 이벤트를 하루씩 모아 통계를 내는 예제로 이 질문들에 답합니다.

이 글에서 알아갈 수 있는 것

매일 자정에 실행하는 것과 3일마다 실행하는 것은 어떻게 설정할까?
Airflow와 데이터를 조회하는 코드는 각각 무엇을 할까? 필요한 구간만 처리하는 증분 처리는 어떻게 설계할까?
어제 데이터는 왜 오늘 처리할까? start_date, 실행 날짜, 실제 시작 시각은 어떻게 다를까?
catchup과 백필은 어떻게 다를까? 기존 태스크를 다시 돌리는 Clear는 언제 쓸까?
같은 날짜를 다시 돌려도 안전하려면 태스크를 어떻게 만들까?

먼저 구분할 것은 Airflow가 실행 일정과 날짜 구간을 제공하고, 태스크가 그 구간으로 실제 데이터를 조회한다는 점입니다. 매일 실행하도록 설정하는 것과 하루치만 가져오도록 조회 조건을 작성하는 것은 각각 해야 합니다.

이번 장의 핵심은 전체 데이터를 매번 다시 처리하는 대신, 각 DAG 실행이 맡은 데이터 구간에서 필요한 부분만 처리하는 증분 파이프라인입니다. 날짜별로 처리 범위를 나누면 반복 작업과 비용을 줄이고, 문제가 생긴 날짜만 다시 처리하기도 쉽습니다.

또한 catchup은 밀린 자동 일정을 따라잡는 정책, 백필은 사용자가 날짜 범위를 지정해 처리하거나 재처리하는 요청입니다. 과거 날짜를 처리한다는 점은 비슷하지만 대상을 정하는 방법이 다릅니다.

3장의 예제는 웹사이트의 사용자 이벤트를 분석하는 파이프라인입니다. 사용자가 어느 페이지를 방문했는지 기록을 가져와 사용자별 통계를 계산합니다. 이번에는 태스크의 순서뿐 아니라 한 번의 실행이 어느 날 데이터를 맡는지가 중요해집니다.

1. 무엇을 만들까: 사용자 이벤트 살펴보기

책 3.1절 · 리스트 3.1

데이터는 예제를 위해 만든 이벤트 API에서 가져옵니다. 이 서비스는 최근 30일 동안의 이벤트만 보관합니다. 더 오래된 기록도 분석하려면, API에서 받은 원본을 직접 저장해 두어야 합니다.

파이프라인은 두 단계입니다.

이벤트 가져오기 → 사용자별 이벤트 수 계산하기
  fetch_events        calculate_stats

앞 태스크가 JSON 파일을 만들고 뒤 태스크가 그 파일을 읽어 CSV 통계를 만듭니다. 처음에는 이벤트를 전부 받아 events.json 하나에 저장하고, 통계도 stats.csv 하나로 만듭니다. 2장의 로켓 사진 예제처럼 앞 태스크의 결과를 뒤 태스크가 읽는 구조입니다.

왜 오퍼레이터를 나눠 태스크를 만들까?

이 예제에서는 데이터를 가져오는 PythonOperator와 통계를 계산하는 PythonOperator를 각각 만들어 두 태스크로 나눕니다. 한 함수에 모두 넣었을 때와 비교하면 다음을 구분해서 관리할 수 있습니다.

나누는 이유	이 예제에서는 어떻게 쓰일까?
역할 분리	`fetch_events`는 원본을 가져와 저장하고, `calculate_stats`는 저장된 원본으로 통계를 계산합니다.
독립적인 실행 관리	두 태스크의 실행 상태·로그·재시도 설정을 따로 관리해 어느 단계가 실패했는지 확인합니다.
선택적 재실행	원본 파일이 있고 통계만 고쳤다면, 데이터를 다시 가져오지 않고 통계 태스크만 돌릴 수 있습니다.
의존성 관리	`fetch_events >> calculate_stats`로 순서를 정합니다. 기본 실행 조건에서는 원본을 만드는 태스크가 성공한 뒤 통계를 계산합니다.

즉, 태스크를 나누는 것은 실패와 재실행을 관리할 단위를 정하는 일입니다. 다만 순서를 연결하는 것과 태스크 사이에 값을 전달하는 것은 별개입니다. 7절의 실습 코드에서 이 차이를 확인하겠습니다.

이 작업을 매일 돌리면 어떨까요? 어제 처리한 이벤트 대부분을 오늘 또 받아 같은 계산을 합니다. 파일도 매번 덮어쓰니, 이대로는 30일보다 오래된 원본을 계속 보관하기 어렵습니다.

그래서 이번 장에서는 두 가지를 바꿉니다. 매일 새로 맡은 하루치만 가져오고, 날짜별 파일에 따로 보관합니다. 먼저 매일 실행하는 설정부터 보겠습니다.

2. 스케줄 걸기: 매일 자정과 3일마다

책 3.2절 · 그림 3.1~3.2 · 표 3.1

스케줄 간격 정의하기 (3.2.1)

2장에서는 schedule_interval="@daily"로 매일 실행하도록 바꿨습니다. 3장에서는 시작 날짜와 종료 날짜까지 함께 보며 스케줄이 어떻게 만들어지는지 살펴봅니다.

# 책의 설정 방식 (Airflow 2.0 기준)
start_date=dt.datetime(2019, 1, 1)
schedule_interval="@daily"

두 설정을 나눠 읽어 보겠습니다. start_date는 자동 실행 일정을 어디서부터 계산할지 정하는 기준, @daily는 매일 자정을 경계로 실행을 만드는 주기입니다. DAG를 켜 두고 스케줄러가 동작하면 다음 날 이후에도 실행이 계속 만들어집니다.

책의 방식에서는 하루가 끝난 뒤 그 하루를 맡은 실행을 만듭니다. 2019년 1월 1일부터 켜 두었다고 하면 다음과 같습니다. 날짜와 시각은 모두 DAG에 설정된 같은 시간대 기준입니다.

실행할 수 있는 가장 이른 시점	이번 실행이 맡은 데이터 구간
1월 2일 00:00	1월 1일 00:00 이상 ~ 1월 2일 00:00 미만
1월 3일 00:00	1월 2일 00:00 이상 ~ 1월 3일 00:00 미만
1월 4일 00:00	1월 3일 00:00 이상 ~ 1월 4일 00:00 미만

이때 매일 1월 1일부터 데이터를 전부 다시 가져오는 것은 아닙니다.

Airflow는 각 실행이 맡을 날짜를 정해 주고, 태스크 코드가 그 날짜를 API 요청이나 SQL 조건에 넣어 하루치만 가져오도록 작성해야 합니다.

@daily만 적고 데이터를 조회하는 코드는 그대로 두면, 매일 같은 전체 데이터를 가져올 수도 있습니다. 이 연결을 3절에서 만듭니다.

또한 start_date를 과거로 적었다고 과거의 모든 실행이 항상 만들어지는 것은 아닙니다. 뒤늦게 DAG를 켰을 때 빠진 날까지 채울지는 catchup 설정이 정합니다. 5절에서 따로 비교하겠습니다.

None과 @once도 구분해야 합니다.

책의 설정	자동 실행	수동 실행
`schedule_interval="@daily"`	매일 반복해서 생성	필요할 때 추가로 가능
`schedule_interval=None`	생성하지 않음	Trigger 버튼이나 API 등으로 요청할 때마다 가능
`schedule_interval="@once"`	한 번 생성	이후에도 필요하면 가능

따라서 None은 자동 실행 일정 없음입니다. 코드를 등록만 하면 한 번도 자동으로 돌지 않습니다.

@once의 “한 번”은 DAG 실행을 자동으로 한 번 만든다는 뜻입니다. 처리할 양과 기간은 태스크 코드가 정합니다. 고객 전체를 옮기는 코드라면 전체를, 특정 날짜를 조회하는 코드라면 그 날짜를 처리합니다. 실패한 태스크의 재시도나 이후 수동 실행도 가능합니다.

@once는 “Airflow 서버를 켤 때마다 한 번”이라는 뜻도 아닙니다. 활성화된 DAG가 시작 날짜 등 실행 조건을 만족하면 자동 실행을 한 번 만듭니다. 실행 기록이 유지되어 있다면 서버를 재시작해도 새로 만들지 않습니다.

end_date는 자동 실행이 맡을 날짜의 마지막 기준입니다. 위 일별 구간 예에서 1월 3일 0시로 설정했다면, 1월 3일 담당 구간까지 대상이고 그 실행은 하루가 끝난 1월 4일 0시 이후에 가능합니다. 따라서 end_date 시각에 이미 돌아가는 태스크를 끄는 설정은 아닙니다. API의 조회 종료 조건과도 구분해야 합니다.

[Airflow 2.0 이후 달라진 점] 스케줄 인자와 하루 데이터 구간

Airflow 3에서는 schedule_interval 대신 schedule을 씁니다. 다만 3.3.2 기본 설정의 schedule="@daily"는 책과 날짜를 다르게 다룹니다. 책처럼 하루 구간이 끝난 뒤 그 구간을 처리하려면 CronDataIntervalTimetable을 명시할 수 있습니다. 이번 실습은 이 방식을 쓰고, 두 방식의 차이는 4절에서 비교합니다. 정기 실행을 끄는 설정도 schedule=None으로 바뀝니다.

Cron 기반의 스케줄 간격 설정하기 (3.2.2)

“매일 몇 시”, “매주 무슨 요일”처럼 달력에 맞추려면 cron 식을 씁니다. 다섯 칸을 순서대로 읽으면 됩니다.

분  시  일  월  요일
0   0   *   *   *     → 매일 자정

Cron 식	의미
`0 * * * *`	매시간 정각
`0 0 * * *`	매일 자정 (`@daily`)
`0 0 * * MON`	매주 월요일 자정
`0 0 1 * *`	매월 1일 자정 (`@monthly`)
`45 23 * * SAT`	매주 토요일 23시 45분
`0 0 * * MON-FRI`	월요일부터 금요일까지 자정
`0 0,12 * * *`	매일 자정과 낮 12시

예를 들어 매주 월요일 자정을 경계로 삼는 데이터 구간은 지난 월요일 자정부터 이번 월요일 자정까지입니다. 주기를 바꾸면 태스크도 그 구간에 맞춰 데이터를 읽어야 합니다. Airflow가 조회 조건까지 자동으로 바꿔 주지는 않습니다.

빈도 기반의 스케줄 간격 설정하기 (3.2.3)

그렇다면 “3일마다”는 어떻게 쓸까요? 0 0 */3 * *로 쓰면 될 것 같지만, 이 식은 매월 날짜 패턴을 고릅니다. 월말에서 다음 달로 넘어갈 때까지 정확히 72시간 간격인 것은 아닙니다.

책에서는 이런 일정한 시간 간격에 timedelta를 사용합니다.

# 책의 설정 방식 (Airflow 2.0 기준)
schedule_interval=dt.timedelta(days=3)

달력의 약속을 따를 때는 cron, 일정한 길이의 구간을 이어 붙일 때는 timedelta라고 나눠 생각하면 됩니다.

[Airflow 2.0 이후 달라진 점] timedelta도 구간 방식을 명시할 수 있습니다

Airflow 3.3.2에서는 DeltaDataIntervalTimetable에 timedelta(days=3)를 넘겨 3일 데이터 구간을 명시할 수 있습니다. 시작을 10월 24일 0시로 잡아 계산했더니 첫 구간은 24~27일이었고, 실행 가능한 시점은 구간이 끝난 27일 0시였습니다. 달력의 특정 시각에 맞추는 CronDataIntervalTimetable과 구분하면 됩니다.

3. 하루치만 가져와 날짜별로 보관하기

책 3.3절 · 그림 3.3~3.4 · 리스트 3.5~3.10

매일 실행하도록 바꿔도, API에서 매번 최근 30일치를 받으면 처리량은 그대로입니다. 이번에는 가져올 범위와 저장할 위치를 함께 바꿉니다.

Airflow와 데이터를 조회하는 코드의 역할

데이터를 조회하는 코드는 API에 요청을 보내 이벤트를 받거나, 데이터베이스에 SQL을 실행해 필요한 행을 읽는 부분입니다. 예를 들어 “24일 0시부터 25일 0시 전까지의 이벤트를 가져와”라는 조건으로 데이터를 받는 코드입니다.

Airflow는 일정·태스크 순서·실패 상태·재시도를 관리하고 실행에 맞는 날짜와 구간을 제공합니다. 실제 조회는 태스크 코드나 조회 기능이 있는 Operator에 연결합니다.

Airflow가 제공·관리하는 것	태스크에서 해야 할 일
매일 실행할 일정	실행되면 수행할 API 호출·SQL 작성
이번 실행의 데이터 구간 시작·끝	그 두 값을 실제 조회 조건에 넣기
태스크 순서·실패 상태·재시도	조회 결과를 계산하고 날짜별로 저장하기

이벤트 데이터 증분 가져오기 (3.3.1)

API가 시작 날짜와 종료 날짜를 받는다고 해 보겠습니다. 10월 24일 하루치의 요청 범위는 다음과 같습니다.

start_date=2026-10-24
end_date=2026-10-25

여기의 start_date와 end_date는 이 예제 API가 받는 조회 조건 이름입니다. 앞 절의 DAG 설정과 이름은 같지만 용도가 다릅니다. DAG의 start_date를 24일로 적었다고 API에 이 값이 저절로 전달되지는 않습니다.

어디에 적는 값인가?	하는 일	매일 바뀔까?
DAG 설정의 `start_date`	자동 실행 일정의 시작 기준	24일로 고정
API 요청의 `start_date`	이번 요청에서 가져올 데이터의 시작	24일, 25일, 26일로 변경
API 요청의 `end_date`	이번 요청에서 가져올 데이터의 끝	25일, 26일, 27일로 변경

이 예제에서 범위는 시작은 포함하고 끝은 제외합니다. 시간으로 풀면 24일 00:00 ≤ 이벤트 시각 < 25일 00:00입니다. 끝을 23:59:59로 잘라 쓰지 않아도 되고, 정확히 25일 0시에 생긴 이벤트는 다음 날 구간에만 들어갑니다. 실제 API나 SQL을 사용할 때도 이 경계 조건을 확인해야 합니다.

원천 데이터가 날짜별 파일로 미리 나뉘어 있어야 하는 것은 아닙니다. 하나의 테이블에 여러 날의 이벤트가 섞여 있어도, 이벤트 시각으로 이 범위를 조회하면 하루치만 고를 수 있습니다. 가져온 결과를 날짜별 파일로 저장하는 것이 뒤에서 설명할 파티셔닝입니다.

SQL이라면 조회 조건은 다음과 같은 모양입니다. :interval_start와 :interval_end는 실행의 구간 값을 바인딩할 자리입니다. 사용하는 데이터베이스 드라이버에 따라 매개변수 표기법은 달라집니다.

SELECT *
FROM events
WHERE event_time >= :interval_start
  AND event_time <  :interval_end;

24일 담당 실행에서는 시작에 24일 0시, 끝에 25일 0시를 넣습니다. 다음 실행에서는 각각 25일 0시와 26일 0시를 넣습니다. API를 쓴다면 같은 두 값을 요청 조건으로 보내고, API 쪽에서도 그 조건으로 데이터를 골라야 합니다. API를 새로 만드는 경우뿐 아니라, 기존 API를 호출하거나 SQL을 작성할 때도 필요한 연결입니다.

매일 실행하는 주기와 하루치만 가져오는 조회 조건은 각각 정해야 합니다. 날짜 조건 없이 전체를 읽는 SQL을 쓰거나 API가 받은 날짜를 무시하면, 매일 실행해도 매번 전체 데이터를 가져올 수 있습니다.

이렇게 각 실행이 맡은 구간만 처리하는 것이 증분 처리입니다. 어제까지의 데이터를 매번 처음부터 계산하는 대신, 이번에 새로 맡은 부분만 처리합니다.

날짜별 데이터 구간과 조회 조건을 연결하면 다음과 같은 설계가 가능합니다.

반복 작업과 비용 줄이기: 매번 전체 데이터를 읽고 계산하는 대신 필요한 범위만 처리하도록 설계할 수 있습니다.
새 데이터 빠르게 반영하기: 새로 맡은 작은 구간부터 처리해 전체 재계산을 기다리는 일을 줄입니다.
날짜별 작업 범위 나누기: 실행 이력과 입력·출력 파일을 같은 날짜 기준으로 확인할 수 있습니다.
오류 구간만 재처리하기: 24일 통계에 문제가 생겼다면 다른 날짜를 건드리지 않고 24일 구간을 다시 처리합니다.

cron처럼 특정 시점에 명령을 실행하는 시스템은 주로 언제 실행할지를 정합니다. 어떤 날짜의 데이터를 처리할지는 명령 안의 코드에서 별도로 정해야 합니다. Airflow의 데이터 구간 방식은 실행 시각과 함께 이번 실행이 맡을 범위를 제공하므로, 며칠 뒤 과거 날짜를 다시 실행해도 그 날짜의 범위로 처리하도록 설계하기 좋습니다.

실행 날짜를 사용하여 동적 시간 참조하기 (3.3.2)

날짜를 코드에 직접 쓰면 매일 파일을 고쳐야 합니다. 그래서 실행마다 Airflow가 제공하는 값을 템플릿에 넣습니다. 책에서는 구간의 시작에 execution_date, 끝에 next_execution_date를 사용합니다. 날짜만 짧게 쓸 때는 ds, next_ds로 표현합니다.

# 책의 템플릿 (Airflow 2.0 기준)
start_date={{ ds }}
end_date={{ next_ds }}

{{ ... }}는 Jinja 템플릿입니다. 태스크가 실행될 때 Airflow가 그 실행에 맞는 값으로 채웁니다. 24일 하루치라면 시작은 24일, 끝은 25일이 되는 식입니다. 날짜를 매번 고치지 않아도 같은 DAG로 다른 날의 데이터를 가져올 수 있습니다.

다만 파이썬 문자열 어디에나 {{ ds }}를 넣으면 자동으로 날짜가 되는 것은 아닙니다. Airflow가 자동으로 템플릿을 처리하는 대상은 오퍼레이터의 template_fields에 지정된 속성입니다. PythonOperator에서는 op_args, op_kwargs, templates_dict가 여기에 해당합니다.

예를 들어 day = "{{ ds }}"라는 일반 변수는 그대로 문자열입니다. 그 값을 PythonOperator의 op_kwargs={"day": day}에 넣으면, 실행 전에 op_kwargs 안의 값이 해당 실행의 날짜로 렌더링되어 함수 인자로 전달됩니다. 원래의 일반 변수 자체가 바뀌는 것은 아닙니다. 이번 실습의 날짜 구간과 파일 경로도 이 방식으로 전달합니다. PythonOperator 템플릿 설명

[Airflow 2.0 이후 달라진 점] 옛 날짜 변수 대신 데이터 구간 변수

Airflow 2.2에서 데이터 구간 개념이 명시적으로 도입됐고, 옛 execution_date는 logical_date라는 이름으로 바뀌었습니다. Airflow 3에서 책의 execution_date, next_execution_date, next_ds를 그대로 쓰면 안 됩니다. 처리할 범위에는 data_interval_start와 data_interval_end를 씁니다. 실행에 붙는 logical_date나 그 날짜 표기인 ds만으로 하루치 구간이 생기는 것은 아닙니다.

하루 구간을 만드는 timetable을 선택했다면, 날짜만 받는 API의 조회 조건은 다음처럼 옮길 수 있습니다.

start_date={{ data_interval_start.in_timezone('Asia/Seoul') | ds }}
end_date={{ data_interval_end.in_timezone('Asia/Seoul') | ds }}

이 코드는 Airflow가 정한 이번 구간을 API의 조회 조건으로 옮기는 연결입니다. 24일을 맡은 실행에서는 start_date=2026-10-24, end_date=2026-10-25가 됩니다. 다음 실행에서는 각각 25일과 26일로 바뀝니다. DAG 설정의 start_date는 계속 24일 그대로입니다.

여기서는 한국 시간으로 바꾼 뒤 날짜 문자열을 만들었습니다. 이 예제 API처럼 시작 포함·끝 제외로 날짜를 해석한다는 전제입니다. 시간 단위 파이프라인이라면 날짜만 남기지 말고 시각까지 전달해야 합니다. 템플릿을 더 자세히 쓰는 방법은 다음 4장에서 다룹니다.

데이터 파티셔닝 (3.3.3)

가져오는 범위만 바꾸고 파일명을 events.json 하나로 유지하면 내일 실행이 오늘 파일을 덮어씁니다. 그러면 오래된 원본을 보관하려던 목적을 이루지 못합니다.

그래서 파일도 날짜별로 나눕니다.

events/2026-10-24.json → stats/2026-10-24.csv
events/2026-10-25.json → stats/2026-10-25.csv
events/2026-10-26.json → stats/2026-10-26.csv

데이터를 관리하기 좋은 조각으로 나누는 것을 파티셔닝이라고 하고, 나뉜 각 조각이 파티션입니다. 여기서는 하루가 파티션 하나입니다.

이제 24일 통계를 다시 계산하려면 24일 JSON만 읽으면 됩니다. 25일 데이터나 26일 결과를 건드릴 이유가 없습니다. 원본을 날짜별로 보관했다면, 외부 API에서 24일 데이터가 사라진 뒤에도 저장해 둔 JSON으로 통계를 다시 계산할 수 있습니다.

다만 재계산할 때 수집부터 다시 실행하면 원본 파일을 빈 응답으로 덮어쓸 수도 있습니다. 보관한 원본으로 재계산할 때는 통계 태스크만 다시 실행하는 식으로 범위를 골라야 합니다. 2장에서 본 Clear의 범위 선택이 여기서도 중요합니다.

4. 어제 데이터는 왜 오늘 처리할까?

책 3.4절 · 그림 3.5~3.7

3장에서 가장 헷갈리는 부분입니다. “10월 24일 실행”이라고 할 때, 실제로 24일에 컴퓨터가 일을 시작했다는 뜻일까요? 아래 날짜는 모두 한국 시간 기준으로 보겠습니다.

책의 구간 처리 방식에서는 24일 데이터를 맡은 실행이라는 뜻입니다. 24일 하루치 데이터를 모두 처리하려면 24일이 끝나야 합니다. 따라서 일을 시작할 수 있는 가장 이른 시점은 25일 0시입니다.

start_date=10월 24일 0시, 하루 데이터 구간, catchup=True이고 스케줄러가 정상적으로 돌고 있다고 해 보겠습니다.

실행이 맡은 날짜	데이터 구간 (한국 시간)	실행 가능한 가장 이른 시점
10월 24일	24일 00:00 이상 ~ 25일 00:00 미만	25일 00:00
10월 25일	25일 00:00 이상 ~ 26일 00:00 미만	26일 00:00
10월 26일	26일 00:00 이상 ~ 27일 00:00 미만	27일 00:00
10월 27일	27일 00:00 이상 ~ 28일 00:00 미만	28일 00:00

Airflow 3.3.2에서도 CronDataIntervalTimetable로 위 네 구간을 확인했습니다. 단, “실행 가능”과 “실제로 태스크가 시작됨”은 다릅니다. 스케줄러 처리, 실행 슬롯, 앞 태스크 등을 기다리면 태스크는 더 늦게 시작합니다. 그래도 맡은 데이터 구간은 바뀌지 않습니다.

시작 날짜, 실행 날짜, 실제 시작 시각은 뭐가 다를까?

start_date는 전체 일정의 시작 기준, logical_date는 개별 실행에 붙은 날짜표라고 생각하면 됩니다. 이 절의 하루 데이터 구간 방식에서는 날짜표가 그 구간의 시작을 가리킵니다. logical_date가 컴퓨터의 현재 날짜라는 뜻은 아닙니다.

따라서 start_date를 적는다고 그날 데이터가 바로 수집되지는 않습니다. 시간표와 catchup 정책으로 실행을 만들고, 그 실행의 태스크가 실제 데이터를 가져옵니다. 오늘 24일 데이터를 재처리한다면 날짜표도 계속 24일입니다.

값 (한국 시간)	24일 데이터를 맡은 실행	25일 데이터를 맡은 실행
DAG 설정의 `start_date`	24일 00:00	24일 00:00으로 그대로
실행의 `logical_date`	24일 00:00	25일 00:00
실행의 `data_interval_start`	24일 00:00	25일 00:00
실행의 `data_interval_end`	25일 00:00	26일 00:00
실제로 시작할 수 있는 가장 이른 시각	25일 00:00	26일 00:00

첫 실행에서는 start_date와 logical_date의 값이 같지만 역할은 다릅니다. 두 번째 실행을 보면 차이가 드러납니다. DAG의 시작 기준은 그대로이고, 이번 실행이 맡은 날짜와 구간만 하루 앞으로 이동합니다. 24일 실행을 며칠 뒤 다시 돌려도 맡은 구간은 24~25일 그대로입니다.

여기서 logical_date가 직접 다음 처리를 하는 것은 아닙니다. 매번 실행되는 것은 같은 태스크 코드이고, 그 코드가 실행마다 달라지는 날짜와 구간을 읽습니다. 이 예제에서는 logical_date를 날짜표로 이해하고, 실제 조회의 양 끝에는 data_interval_start와 data_interval_end를 사용하면 됩니다.

DAG의 start_date로 일정의 시작 기준 정하기
  → 시간표가 각 실행의 날짜표와 데이터 구간 계산
  → 태스크 코드가 이번 구간을 API·SQL 조회 조건에 넣기
  → 조회한 데이터를 해당 날짜 파일에 저장

또한 데이터 구간이 끝났다고 원천 데이터가 반드시 모두 도착했다는 뜻은 아닙니다. 늦게 들어오는 이벤트나 지연 적재가 있다면 별도로 기다리거나 해당 파티션을 다시 계산하는 정책이 필요합니다. Airflow가 시간표만 보고 원천 데이터의 완성까지 보장해 주지는 않습니다.

매일 깨우는 일정과 하루치 구간을 함께 정하는 일정

위 설명을 Airflow 3에서도 그대로 쓰려면 timetable, 즉 실행 시점과 데이터 구간을 계산하는 시간표를 골라야 합니다. CronDataIntervalTimetable은 이름을 나눠 보면 뜻이 드러납니다.

Cron: 매일 자정처럼 달력의 시각을 경계로 삼습니다.
DataInterval: 이전 경계부터 이번 경계까지를 한 실행의 데이터 구간으로 정합니다.
Timetable: 그 구간이 끝난 뒤 실행할 수 있도록 일정을 계산합니다.

매일 자정을 경계로 고르면 24일 0시부터 25일 0시까지가 한 구간이고, 25일 0시 이후에 그 구간을 처리합니다. 이 시간표가 API를 직접 호출하거나 날짜별 파일을 찾아 주는 것은 아닙니다. 계산해 준 시작·끝을 태스크가 조회 조건으로 사용해야 합니다.

[Airflow 2.0 이후 달라진 점] 매일 실행하기와 하루 데이터 구간 만들기를 구분합니다

Airflow 3.3.2 기본 설정의 schedule="@daily"는 “매일 자정에 실행을 만든다”는 방식입니다. 여기에 어제 하루치 구간이 자동으로 붙지는 않습니다. 책처럼 “하루가 끝나면 그 하루를 맡긴다”는 방식을 원하면 CronDataIntervalTimetable을 명시할 수 있습니다. 둘 다 매일 실행할 수 있지만 실행에 붙는 데이터 구간이 다릅니다.

하루 구간을 명시하는 Airflow 3 코드는 다음과 같습니다. 시간대도 함께 정해 한국의 자정을 기준으로 잡았습니다.

import pendulum
from airflow.sdk import DAG, CronDataIntervalTimetable

with DAG(
    dag_id="daily_events",
    start_date=pendulum.datetime(2026, 10, 24, tz="Asia/Seoul"),
    schedule=CronDataIntervalTimetable("0 0 * * *", timezone="Asia/Seoul"),
    catchup=False,
):
    ...  # 태스크는 7절에서 채웁니다.

여기서 start_date는 “24일부터 일정 계산”, CronDataIntervalTimetable은 “매일 자정으로 나눈 하루 구간”, catchup=False는 “처음 켤 때 밀린 구간은 모두 만들지 않음”입니다. 예를 들어 28일 오전에 이 DAG를 처음 켜면 24일부터 차례로 시작하는 것이 아니라, 가장 최근에 끝난 27일 구간부터 처리합니다. 이후 29일 자정에는 28일 구간, 30일 자정에는 29일 구간으로 계속 이어집니다. 24~26일도 필요하면 5절의 catchup이나 백필을 사용합니다.

기본 @daily로도 DAG와 태스크는 정상적으로 실행됩니다. 매일 API에서 현재 상품 목록을 받아 갱신하는 작업은 하루 구간 없이도 할 수 있습니다. 코드에서 조회 범위를 직접 계산하면 일별 처리도 가능합니다. 이번 예제는 Airflow가 제공하는 하루 구간을 그대로 조회에 사용하려고 CronDataIntervalTimetable을 명시합니다.

@daily에 하루 구간도 붙일지 정하는 설정

scheduler.create_cron_data_intervals는 @daily라고 짧게 적었을 때 하루 데이터 구간도 만들어 줄지 정하는 기본 선택입니다. 매일 실행할지 말지를 켜고 끄는 설정이 아닙니다.

False(3.3.2 기본값)이면 실행 시점 중심의 CronTriggerTimetable, True이면 구간을 제공하는 CronDataIntervalTimetable을 고릅니다. 이 선택은 schedule에 cron 문자열을 적었을 때 적용됩니다. 시간표 객체를 직접 적으면 그 시간표를 사용하므로, 이번 실습에서는 환경 설정을 바꾸지 않습니다.

이미 매일 돌고 있는 DAG의 10월 28일 자정 정기 실행을 예로 들면 더 분명합니다. 아래 시각은 모두 한국 시간이고, 기본 @daily에는 별도 구간을 지정하지 않았다고 가정합니다.

태스크에 전달되는 값	`False`: 실행 시점 중심	`True`: 하루 데이터 구간 제공
실행 가능한 시점	28일 00:00	28일 00:00
`logical_date`	28일 00:00	27일 00:00
`data_interval_start`	28일 00:00	27일 00:00
`data_interval_end`	28일 00:00	28일 00:00

False는 “28일 자정이니 태스크를 실행해”입니다. True는 “28일 자정이니 태스크를 실행하고, 이번 실행에는 27일 0시부터 28일 0시까지라는 범위를 붙여 줘”입니다. True가 어제 실행 하나를 추가로 돌리거나, 어제 데이터를 직접 조회해 준다는 뜻은 아닙니다. 두 경우 모두 위 표의 자정 경계에 대응하는 실행 하나를 비교하고 있습니다.

이번 실습에서는 시간표가 제공한 27~28일 구간을 태스크가 조회 조건에 넣어 27일 하루치를 가져옵니다.

날짜 변수 이름만 바꾸면 왜 부족할까?

3절에서는 책의 ds·next_ds를 data_interval_start·data_interval_end로 옮겼습니다. 하지만 새 변수는 이미 계산된 구간을 읽는 이름입니다. 변수 이름을 바꾸는 것 자체가 하루 구간을 만들어 주지는 않습니다.

위 비교에서 False의 구간 값을 API 조건에 그대로 넣으면 시작도 2026-10-28, 끝도 2026-10-28입니다. 이 예제의 시작 포함·끝 제외 조건으로는 빈 범위입니다. 반면 True의 값은 시작 2026-10-27, 끝 2026-10-28이어서 27일 하루치가 됩니다.

따라서 하루 구간을 만드는 timetable을 선택하고, 태스크는 그 구간 변수를 조회 조건으로 사용해야 합니다. 빈 범위가 됐다는 것은 조회할 하루치가 전달되지 않았다는 뜻이며, 기본 @daily 자체가 실행되지 않는다는 뜻은 아닙니다.

수동 실행도 따로 봐야 합니다. UI의 Trigger 버튼으로 만든 실행은 정기 실행과 구간이 결정되는 경로가 다를 수 있습니다. 버튼을 누르면서 논리적 날짜를 지정했다고 그 날짜가 곧 처리 구간의 시작이라고 가정하면 안 됩니다. 하루치 파티션을 확인하는 이번 실습에서는 정기 실행과 날짜 범위를 지정하는 백필을 사용합니다.

Airflow 화면이나 CLI에서 2026-10-26 15:00 UTC로 보인다면 한국 시간으로는 27일 00:00입니다. 시간대를 바꿔 읽어야 하므로 날짜 숫자만 비교하지 않도록 주의해야 합니다.

Airflow 3의 기본 시간표 선택은 공식 설정 문서, 구간과 timetable 차이는 공식 Timetables 문서에서도 확인할 수 있습니다.

5. 빠진 날짜를 채우기: catchup과 백필

책 3.5절 · 그림 3.8 · 리스트 3.11

DAG의 시작을 24일로 적고, 실제로는 28일 오전 10시에 처음 켰다고 해 보겠습니다. 지금까지 끝난 하루 구간은 24·25·26·27일, 네 개입니다. 이 네 개를 모두 처리할까요, 가장 최근 것만 처리할까요?

이때 자동 스케줄의 정책을 정하는 것이 catchup입니다. 앞 절과 같은 하루 데이터 구간 방식을 기준으로 비교하면 다음과 같습니다.

설정	처음 활성화했을 때 처리 대상
`catchup=True`	24·25·26·27일의 완료된 구간 네 개
`catchup=False`	가장 최근에 끝난 27일 구간 하나

catchup=True이면 이전 자동 실행 다음부터 현재까지 끝난 일정을 따라 날짜별 DAG 실행을 만듭니다. 처음 켰고 이전 실행이 없다면 start_date를 기준으로 시작합니다. 기준은 원본 데이터나 JSON 파일의 존재 여부가 아니라 Airflow에 기록된 실행 일정과 실행 이력입니다.

새로 만들어진 실행에서는 DAG에 정의한 태스크들을 의존 순서와 실행 조건에 따라 진행합니다. 이번 예제라면 fetch_events로 해당 날짜 데이터를 가져오고, 이어서 calculate_stats로 통계까지 계산합니다. 데이터를 가져오는 태스크만 따로 실행하는 기능이 아닙니다.

위 예에서는 아래 네 실행을 만들고, 각 실행이 자기 하루치를 가져옵니다. 각 줄의 끝 날짜는 제외합니다.

24일 담당 실행 → 24일 00:00 이상 ~ 25일 00:00 미만 조회
25일 담당 실행 → 25일 00:00 이상 ~ 26일 00:00 미만 조회
26일 담당 실행 → 26일 00:00 이상 ~ 27일 00:00 미만 조회
27일 담당 실행 → 27일 00:00 이상 ~ 28일 00:00 미만 조회

데이터 구간은 “실행 하나가 어느 범위를 맡는가”, catchup은 “밀린 과거 실행도 만들 것인가”를 정합니다. 서로 다른 질문입니다. 하루 구간을 제공해도 catchup=False이면 위 예에서는 마지막 27일 실행만 만들고, catchup=True이면 네 실행을 모두 만듭니다. 네 실행을 반드시 하나씩 순서대로 돌린다는 뜻은 아니며, 실행 슬롯과 동시 실행 제한에 따라 진행됩니다.

28일 하루는 아직 끝나지 않았으므로 이 표에는 없습니다. catchup=False여도 다음 날 이후의 정기 실행은 계속 만들어집니다. “한 번만 실행”한다는 뜻은 아닙니다.

책에서는 이런 과거 구간 채우기를 백필이라는 큰 개념으로 설명합니다. 지금 Airflow를 사용할 때는 스케줄러의 자동 따라잡기인 catchup과 범위를 지정해 요청하는 Backfill 기능을 나눠 생각하면 편합니다.

[Airflow 2.0 이후 달라진 점] catchup 기본값과 백필 명령

Airflow 3에서는 catchup_by_default 기본값이 False입니다. 책처럼 모든 과거 구간을 자동으로 만들고 싶으면 catchup=True를 명시합니다. 명령줄 백필도 옛 airflow dags backfill 대신 airflow backfill create를 사용합니다. 이번 실습은 catchup=False로 최신 구간을 한 번 확인한 뒤, 필요한 과거 세 날짜만 백필합니다.

catchup과 백필은 같은 일을 하는 걸까?

과거 날짜별로 DAG 실행을 만들어 같은 태스크들을 진행한다는 점은 비슷합니다. 차이는 실행을 만들도록 요청하는 방법과 대상 선택에 있습니다.

구분	catchup	Backfill
누가 대상을 정할까?	스케줄러가 다음 자동 일정부터 따라잡음	사용자가 처리할 날짜 범위를 지정
기존 실행이 없는 날짜	따라잡을 자동 일정에 해당하면 생성	요청한 범위에 있으면 생성 가능
이미 성공·실패한 날짜	catchup만 켠다고 새로 다시 만들지 않음	재처리 정책을 정해 새 실행을 만들 수 있음
계속 유지하는 정책일까?	DAG의 자동 일정 정책	필요한 때 요청하는 별도 작업

28일에 처음 켠 DAG에서 24~27일을 모두 자동으로 채우려면 catchup=True를 사용할 수 있습니다. 반면 catchup=False로 운영하다가 “24~26일만 채워 줘”라고 요청할 때도 백필을 쓸 수 있습니다. 백필이 이미 처리한 날짜 전용인 것은 아닙니다. catchup=False로 과거를 건너뛰고 최근 실행까지 이미 만들었다면, 나중에 True로 바꾸는 것만으로 그보다 앞선 빈 날짜들을 모두 채우지는 않습니다. 그런 날짜는 범위를 정해 백필합니다.

이미 성공한 25·26일의 계산식을 고쳤다면, 백필에서 완료된 날짜도 재처리하도록 정책을 선택할 수 있습니다. 특정 요일의 데이터만 다시 필요하다면 해당 날짜들을 골라 요청합니다. 구분 기준은 요일이 아니라 자동으로 따라잡을지, 내가 범위를 정해 요청할지입니다.

또한 실패한 실행이 이미 있으면 catchup을 켜는 것만으로 같은 날짜를 새로 실행하지는 않습니다. 기존 태스크를 Clear하거나, 백필의 재처리 정책을 선택합니다.

기존 실행의 일부만 다시 돌리고 싶다면?

예를 들어 24~26일 전체 파이프라인을 돌리려면 백필을 요청할 수 있습니다. 24일 원본 JSON은 이미 있고 통계만 고쳤다면, 해당 날짜의 calculate_stats를 Clear하는 편이 목적에 맞습니다.

Airflow 3의 백필은 날짜 범위와 기존 실행을 다시 처리할지 정하는 정책을 함께 받습니다. 7절에서 대상 날짜를 먼저 확인한 뒤 실제로 실행해 보겠습니다.

과거 실행을 만든다고 원본 데이터가 생기지는 않습니다

예제의 외부 API는 최근 30일치만 줍니다. 60일 전 날짜로 실행을 만들어도 API에 데이터가 없다면 받을 수 없습니다. Airflow는 과거 실행의 날짜와 순서를 관리할 수 있지만, 원천에서 사라진 데이터까지 복구하지는 못합니다.

그래서 3절의 날짜별 원본 보관이 중요합니다. 백필할 수 있는 코드와 다시 읽을 수 있는 원본 데이터가 함께 있어야 과거를 재처리할 수 있습니다.

6. 다시 돌려도 믿을 수 있는 태스크 만들기

책 3.6절 · 그림 3.9~3.10 · 리스트 3.12~3.14

2장에서는 실패한 태스크를 다시 돌리는 방법을 봤습니다. 이번에는 날짜별 백필까지 할 수 있게 됐습니다. 그렇다면 마지막 질문이 남습니다. 다시 실행하면 결과도 안전할까요?

Airflow가 재실행을 관리해 준다고 해서 태스크 안에서 만든 파일이나 외부 시스템의 상태까지 자동으로 되돌려 주지는 않습니다. 그래서 3장은 원자성과 멱등성으로 끝납니다.

원자성 (3.6.1): 실패했을 때 반쪽 결과를 내놓지 않기

파일을 쓰다가 실패했다고 해 보겠습니다. 최종 파일에 바로 한 줄씩 쓰고 있었다면, 이름은 정상인데 내용은 절반뿐인 파일이 남을 수 있습니다. 다음에 이 파일을 읽는 사람은 완성된 결과인지 구분하기 어렵습니다.

로컬 파일에서는 다음처럼 쓸 수 있습니다.

임시 파일에 끝까지 쓰기 → 다 썼으면 최종 파일로 교체
                  실패 → 기존 최종 파일 유지

이렇게 결과를 내놓는 경계를 정하면, 다운스트림에는 완성된 결과만 보이도록 만들 수 있습니다. 7절의 실습은 같은 디렉터리에 임시 파일을 쓰고 os.replace()로 교체합니다. 이 설명은 같은 로컬 파일시스템에서 파일 하나를 교체하는 경우입니다. 여러 파일을 한꺼번에 바꾸거나 데이터베이스까지 갱신하는 전체 작업을 하나의 트랜잭션으로 만들어 주는 코드는 아닙니다.

원자성을 생각하며 태스크를 나눌 때는 각 태스크가 하나의 완결된 결과를 만들도록 경계를 잡습니다. 결과를 완성해 공개하는 단계와 그 결과로 다음 일을 하는 단계를 구분하면, 실패했을 때 어디까지 끝났고 어디서부터 다시 해야 하는지 분명해집니다. 각 태스크 안에서도 반쪽 결과가 공개되지 않도록 파일 교체나 트랜잭션을 설계해야 합니다.

3.6.1절은 통계 계산과 이메일 발송을 한 태스크 안에 넣는 예로 태스크 경계도 설명합니다. 통계 파일까지 다 만들었는데 이메일에서 실패하면, 재시도할 때 통계도 다시 계산해야 합니다.

calculate_stats → send_stats

두 태스크로 나누면 발송만 실패했을 때 send_stats만 다시 돌릴 수 있습니다. 다만 태스크를 나누기만 하면 모든 부수 효과가 원자적으로 바뀌는 것은 아닙니다. 이메일 서버가 메일을 받은 직후 연결이 끊기면, 보내는 쪽에서는 실패로 판단해도 실제로는 메일이 발송됐을 수 있습니다. 이런 경우는 중복 방지까지 별도로 생각해야 합니다.

그렇다고 작은 동작을 무조건 모두 나누는 것도 답은 아닙니다. API에 로그인해서 받은 임시 토큰으로 바로 조회해야 한다면, 인증과 조회를 한 태스크 안에서 처리하는 편이 자연스러울 수 있습니다. 실패했을 때 어디서부터 다시 해야 의미가 맞는지를 기준으로 나눕니다.

멱등성 (3.6.2): 같은 입력으로 다시 돌려도 결과가 같기

24일 이벤트 세 건을 파일에 추가하는 태스크를 생각해 보겠습니다. 재시도할 때마다 같은 세 건을 덧붙이면 세 건이 여섯 건, 아홉 건이 됩니다. 성공 횟수에 따라 데이터가 달라집니다.

반대로 events/2026-10-24.json을 동일한 세 건으로 교체하면, 몇 번 실행해도 최종 내용은 같습니다. 이것이 멱등성입니다.

저장 방식	같은 입력으로 세 번 실행하면
같은 파일에 이벤트를 계속 추가	이벤트가 중복될 수 있음
해당 날짜 파티션을 완성된 결과로 교체	최종 결과가 같음
데이터베이스에 매번 새 행 삽입	같은 키의 데이터가 중복될 수 있음
정해진 키로 upsert하거나 대상 파티션을 교체	중복을 피하도록 설계할 수 있음

앞에서 날짜 파티션을 만든 이유가 여기까지 이어집니다. 실행이 맡는 데이터 범위와 결과를 교체할 범위를 같게 잡으면, 다른 날짜를 건드리지 않고 재처리할 수 있습니다.

단, “같은 날짜”가 반드시 “같은 입력”인 것은 아닙니다. 원천 API가 과거 데이터를 수정하면 다시 받은 내용도 달라집니다. 결과를 정확히 재현하려면 어떤 버전의 원본을 읽을지도 정해야 합니다. 이번 실습에서는 날짜 구간마다 같은 샘플을 만들어 입력을 고정했습니다.

7. 실습: 하루치 이벤트를 처리하는 Airflow 3 DAG

저자 저장소 chapter03 · 리스트 3.8~3.10의 날짜별 처리 흐름을 바탕으로 직접 작성

원본 예제는 저자 저장소의 chapter03에 있습니다. 날짜별 파일을 만드는 08_templated_path.py의 설정과 템플릿에는 다음 표현이 나옵니다.

# 책 예제 중 설정·템플릿 일부 (서로 다른 줄을 발췌)
schedule_interval="@daily"
"/data/events/{{ds}}.json"
"start_date={{ds}}&end_date={{next_ds}}"

이 코드를 Airflow 3로 옮길 때는 import와 인자 이름뿐 아니라 어떤 데이터 구간이 만들어지는지도 바꿔야 합니다.

책 예제	이번 실습에서 바꾼 내용
`from airflow import DAG`	공개 경로인 `from airflow.sdk import DAG`
`airflow.operators.python`	`airflow.providers.standard.operators.python`
`schedule_interval="@daily"`	`schedule=CronDataIntervalTimetable(...)`
시간대 없는 시작 날짜	`pendulum.datetime(..., tz="Asia/Seoul")`
`ds`, `next_ds`로 처리 범위 결정	`data_interval_start`, `data_interval_end`
기본값에 맡긴 catchup	`catchup=False` 명시 후 필요한 범위만 백필
별도 이벤트 API와 pandas	고정된 샘플 이벤트와 표준 라이브러리로 실습 구성
최종 파일에 바로 쓰기	임시 파일을 완성한 뒤 해당 날짜 파일 교체

마지막 두 줄은 Airflow 호환성 때문에 필수인 변경은 아닙니다. 별도 API 서버 없이 스케줄·파티션·재실행을 확인하고, 6절에서 다룬 저장 방식을 코드로 보여 주기 위한 선택입니다. 실제 서비스 데이터를 수집하는 예제가 아니라는 점은 구분해 두겠습니다.

1·2장에서 사용한 ruff도 확인합니다.

uvx ruff@latest check --select AIR3 dags/

다만 ruff가 통과했다고 날짜 의미까지 맞다는 뜻은 아닙니다. 옛 import와 삭제된 인자는 찾을 수 있어도, 이 파이프라인에 하루 데이터 구간이 필요한지는 코드의 목적과 실행 결과로 확인해야 합니다.

Airflow 3 코드

로컬 저장소의 dags/ch03_daily_events.py입니다. 원본을 복사하지 않고, 이번 장의 처리 흐름을 확인하도록 직접 작성했습니다.

import csv
import io
import json
import os
import tempfile
from collections import Counter
from pathlib import Path

import pendulum

from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG, CronDataIntervalTimetable

TZ = "Asia/Seoul"
# 실행한 오늘 날짜가 아니라, 담당 구간의 시작 날짜로 파일을 나눕니다.
PARTITION = "{{ data_interval_start.in_timezone('Asia/Seoul') | ds }}"
EVENTS_PATH = f"/tmp/ch03/events/{PARTITION}.json"
STATS_PATH = f"/tmp/ch03/stats/{PARTITION}.csv"

def _atomic_write(output_path, text):
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary_path = None
    try:
        # 같은 파일시스템의 임시 파일에 끝까지 쓴 뒤 최종 파일을 교체합니다.
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", dir=path.parent, delete=False
        ) as temporary:
            temporary_path = Path(temporary.name)
            temporary.write(text)
        os.replace(temporary_path, path)
    finally:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)

def _fetch_events(start, end, output_path):
    start, end = pendulum.parse(start), pendulum.parse(end)
    # 외부 API 대신 샘플 5건을 준비합니다. 구간 안은 3건, 구간 밖은 2건입니다.
    samples = [
        (start.subtract(hours=1), "outside"),
        (start.add(hours=8), "alice"),
        (start.add(hours=12), "alice"),
        (start.add(hours=13), "bob"),
        (end, "outside"),
    ]
    # 실제 API·SQL 조회에도 이 시작 포함·끝 제외 조건을 적용해야 합니다.
    events = [
        {"timestamp": timestamp.isoformat(), "user": user}
        for timestamp, user in samples
        if start <= timestamp < end
    ]
    _atomic_write(output_path, json.dumps(events, indent=2))
    print(f"interval=[{start.isoformat()}, {end.isoformat()}) events={len(events)}")
    print(f"wrote {output_path}")

def _calculate_stats(input_path, output_path):
    events = json.loads(Path(input_path).read_text(encoding="utf-8"))
    counts = Counter(
        (pendulum.parse(event["timestamp"]).in_timezone(TZ).to_date_string(), event["user"])
        for event in events
    )
    buffer = io.StringIO(newline="")
    writer = csv.writer(buffer, lineterminator="\n")
    writer.writerow(["date", "user", "events"])
    writer.writerows((date, user, count) for (date, user), count in sorted(counts.items()))
    _atomic_write(output_path, buffer.getvalue())
    print(buffer.getvalue(), end="")
    print(f"wrote {output_path}")

with DAG(
    dag_id="daily_events",
    # 자동 일정의 시작 기준입니다. 실제 조회 범위는 실행마다 전달받습니다.
    start_date=pendulum.datetime(2026, 10, 24, tz=TZ),
    # 한국 시간 하루 구간을 만들고, 그 하루가 끝난 뒤 실행합니다.
    schedule=CronDataIntervalTimetable("0 0 * * *", timezone=TZ),
    # 처음 켤 때 최신 완료 구간만 처리합니다. 과거 날짜는 뒤에서 백필합니다.
    catchup=False,
    max_active_runs=1,
    default_args={"retries": 1},
    tags=["book", "chapter03"],
):
    fetch_events = PythonOperator(
        task_id="fetch_events",
        python_callable=_fetch_events,
        op_kwargs={
            # Airflow의 구간 값을 조회 함수의 start·end 인자로 전달합니다.
            "start": "{{ data_interval_start.isoformat() }}",
            "end": "{{ data_interval_end.isoformat() }}",
            "output_path": EVENTS_PATH,
        },
    )
    calculate_stats = PythonOperator(
        task_id="calculate_stats",
        python_callable=_calculate_stats,
        op_kwargs={"input_path": EVENTS_PATH, "output_path": STATS_PATH},
    )
    fetch_events >> calculate_stats

코드 읽는 법

같은 날짜의 재실행: 샘플 입력과 CSV 정렬 순서를 고정하고 해당 파티션을 교체하므로, 같은 날짜를 다시 처리하면 같은 내용이 나옵니다.
실패 시 동작: _atomic_write에서 오류가 나면 예외가 태스크 밖으로 전달되어 Airflow도 실패를 알 수 있습니다. 남은 임시 파일은 정리합니다.
로컬 실습의 파일 공유: 컨테이너 하나에서 두 태스크가 /tmp/ch03을 함께 읽습니다. 여러 워커 환경에서는 공용 오브젝트 스토리지 같은 저장소가 필요합니다.

이 코드의 파티션 파일명은 하루 구간에 맞춘 것입니다. 시간별 구간으로 바꾼다면 파일명도 시각을 포함하도록 바꿔야 같은 날의 여러 실행이 서로 덮어쓰지 않습니다.

태스크 순서를 연결하면 반환값도 전달될까?

fetch_events >> calculate_stats는 앞 태스크 다음에 뒤 태스크를 실행하라는 의존성입니다. 태스크 A의 return 값이 태스크 B의 op_kwargs에 자동으로 들어가지는 않습니다. 이번 코드도 두 태스크의 파일 경로를 직접 맞춰 두었기 때문에, 앞에서 만든 JSON을 뒤에서 읽습니다. 앞 함수의 반환값을 넘기는 구조는 아닙니다.

태스크 사이에 파일 경로·식별자·건수 같은 작은 값을 전달할 때는 XCom을 사용할 수 있습니다. PythonOperator는 기본 설정에서 None이 아닌 반환값을 XCom의 return_value로 저장하지만, 뒤 태스크가 그 값을 사용하도록 참조하거나 꺼내는 연결은 별도로 지정해야 합니다. 이벤트 전체처럼 큰 데이터는 저장소에 두고 작은 참조 정보만 전달하는 식으로 나눕니다. XCom 공식 설명

직접 돌려 보기

실습 환경은 앞선 글과 같은 vividbaek/airflow-book-study 저장소입니다. 위 코드를 dags/ch03_daily_events.py로 저장한 뒤 실행합니다.

docker compose up

http://localhost:8080에서 daily_events DAG를 찾습니다. catchup=False이므로 처음에는 가장 최근에 끝난 하루 구간을 처리합니다. 2026년 10월 28일에 확인했을 때는 27일 파티션 하나가 생겼습니다. 읽는 날짜가 다르면 첫 파티션 날짜도 달라집니다.

먼저 백필할 날짜를 확인합니다. +09:00을 붙여 한국 시간 자정임을 명확히 하고, --dry-run으로 실행 대상만 봅니다.

docker compose exec airflow airflow backfill create \
  --dag-id daily_events \
  --from-date '2026-10-24T00:00:00+09:00' \
  --to-date '2026-10-26T00:00:00+09:00' \
  --reprocess-behavior none \
  --dry-run

이 DAG에서 대상은 논리적 날짜 24·25·26일이고, 각각 24~25일, 25~26일, 26~27일 데이터 구간입니다. 종료 인자에 26일을 썼다고 마지막 데이터 구간도 26일 0시에 끝나는 것은 아닙니다.

대상이 맞으면 --dry-run을 빼고 실행합니다.

docker compose exec airflow airflow backfill create \
  --dag-id daily_events \
  --from-date '2026-10-24T00:00:00+09:00' \
  --to-date '2026-10-26T00:00:00+09:00' \
  --reprocess-behavior none \
  --max-active-runs 1

none은 해당 논리적 날짜에 실행이 이미 있으면 새로 만들지 않는 정책입니다. failed는 기존 실행이 실패한 경우에도, completed는 성공하거나 실패해 완료된 경우에도 다시 만들 수 있게 합니다. 최신 실행이 아직 대기·실행 중이면 새로 만들지 않습니다. 재처리 정책은 공식 Backfill 문서에 정리되어 있습니다. 이번에는 --max-active-runs 1로 한 번에 실행할 백필 수도 정했습니다.

끝나면 결과를 봅니다.

docker compose exec airflow ls /tmp/ch03/events
docker compose exec airflow cat /tmp/ch03/stats/2026-10-24.csv

24일 결과는 다음과 같습니다.

date,user,events
2026-10-24,alice,2
2026-10-24,bob,1

실제로 돌린 결과

Airflow 3.3.2 공식 Docker 이미지에서 2026년 10월 28일에 확인했습니다.

정기 실행: DAG를 등록하자 가장 최근 완료 구간인 27~28일을 맡은 실행 하나가 생겼고, 두 태스크가 모두 성공했습니다. 한국 시간의 논리적 날짜는 27일 0시였습니다.
과거 세 날짜 백필: 24·25·26일 실행 세 개가 추가됐고 모두 성공했습니다. 각 날짜의 JSON과 CSV가 따로 만들어졌습니다.
구간 경계: 날짜마다 샘플 다섯 건 중 구간 안의 세 건만 JSON에 남았습니다. 종료 경계의 이벤트는 들어가지 않았습니다. CSV에는 Alice 2건과 Bob 1건이 기록됐습니다.
통계만 다시 실행: 24일의 calculate_stats만 Clear해 재실행했습니다. fetch_events의 시도 횟수는 그대로였고, 통계 태스크의 시도 횟수만 늘었습니다. 재실행 전후 JSON·CSV의 SHA-256을 비교했을 때 모든 파일 내용이 같았습니다.
파일 공개 직전 실패: 임시 파일을 최종 파일로 바꾸는 부분에 의도적으로 오류를 주는 별도 함수 검사에서는, 기존 최종 파일 내용이 유지되고 임시 파일도 정리됐습니다. 이 검사는 함수 단위이며, 컨테이너 강제 종료나 스토리지 장애 전체를 재현한 것은 아닙니다.

이 실습은 실제 외부 API의 보관 기간이나 지연 도착을 재현하지는 않습니다. 구간이 어떻게 정해지고, 그 구간이 파일명과 재실행 범위까지 어떻게 이어지는지를 확인하기 위한 예제입니다.

8. 3장을 마치며

처음에 던진 다섯 가지 질문에 답해 보면 이렇습니다.

달력의 특정 시각은 cron, 일정한 길이의 구간은 timedelta로 표현합니다. Airflow 3에서 데이터 구간을 처리하려면 그 목적에 맞는 timetable도 명시합니다.
Airflow가 제공한 데이터 구간을 조회 조건으로 사용하면 필요한 범위만 증분 처리하도록 설계할 수 있습니다. 전체 데이터의 반복 조회·계산을 줄이고, 날짜별 원본과 결과를 보관하면 오류가 난 구간만 재처리하기도 쉽습니다. 실제 데이터를 가져오는 일은 태스크가 담당합니다.
하루치 데이터는 하루가 끝나야 처리할 수 있습니다. 구간 방식에서 논리적 날짜는 맡은 구간의 시작이고, 실제 실행은 구간 끝 이후입니다. start_date는 스케줄 계산의 기준이므로 실제 태스크 시작 시각과 구분합니다.
catchup은 자동으로 빠진 구간을 따라잡는 정책, 백필은 지정한 날짜 범위를 처리하는 요청, Clear는 기존 태스크의 재실행입니다. 과거 실행을 만들 수 있어도 원본 데이터가 없으면 복구할 수 없습니다.
실패했을 때 반쪽 결과를 공개하지 않는 원자성과, 같은 입력을 다시 처리해도 결과가 같은 멱등성이 필요합니다. 태스크를 완결된 작업 단위로 나누고, 각 태스크 안에서도 결과가 안전하게 공개되도록 설계합니다.

3장을 읽고 나면 DAG의 날짜를 “언제 컴퓨터가 일을 시작했는가”와 “어느 구간의 데이터를 맡았는가”로 나눠 볼 수 있습니다. 이 구분이 파일명, 증분 처리, 백필, 재실행을 하나로 이어 줍니다. 전체 데이터를 매번 다시 처리하지 않고 실행이 맡은 범위만 처리하며, 그 결과를 안전하게 교체하는 파이프라인이 이번 장의 핵심입니다.

다음 4장(Airflow 콘텍스트를 사용하여 태스크 템플릿 작업하기)에서는 이번에 쓴 {{ ... }}가 언제 어떤 값으로 채워지는지, 오퍼레이터에서 무엇을 템플릿으로 만들 수 있는지 더 자세히 다룹니다.

이 글은 Apache Airflow 한국 사용자 모임 운영진으로 활동하며 《Apache Airflow 기반의 데이터 파이프라인》을 학습하고, Airflow 3에서 달라진 내용을 더해 정리한 기록입니다. 글을 읽으시며 의견이나 함께 나누고 싶은 인사이트가 있다면 언제든 댓글로 남겨주세요!