Airflow 태스크, 재시도와 backfill로 다시 돌려도 안전할까?

이번 시간에는 Docker로 띄운 Airflow와 PostgreSQL에서, INSERT만 하는 태스크가 재시도·clear·backfill을 거치며 결과를 얼마나 늘리는지 재 보고, 다시 실행해도 결과가 같은 쓰기 방법과 비교하며 배워가는 시간입니다.

예를 들어서 매일 0시에 전날 매출을 집계하는 Airflow DAG가 있다고 해볼게요. 어느 날 DB 연결이 잠깐 끊겨 태스크가 실패했고, Airflow가 설정대로 한 번 더 실행해서 성공했어요. 그런데 다음 날 아침 대시보드의 전날 매출이 평소의 두 배였습니다. 왜 그럴까요?

실패한 첫 시도가 사실은 집계 결과를 이미 저장해 둔 상태였고, 재시도가 같은 결과를 한 번 더 넣은 거예요.

Airflow는 같은 태스크를 여러 번 실행할 수 있어요. 실패하면 재시도하고, 사람이 clear하면 다시 실행하고, backfill로 지난 날짜를 다시 처리하기도 하죠. 그래서 태스크는 몇 번을 실행해도 결과가 한 번 실행한 것과 같아야 해요. 이런 성질을 멱등성이라고 합니다.

실습 준비

이 글의 명령은 모두 실습 코드의 airflow-idempotent-dag/ 디렉터리에서 실행합니다.

• compose.yaml: PostgreSQL 17과 Airflow 3.3.2 standalone 컨테이너 정의
• init.sql: airflow·warehouse DB, 원본 주문 테이블, 결과 테이블 네 개를 만드는 초기화 SQL
• dags/sales.py: 쓰기 방식별 집계 DAG
• report.sql: 세 결과 테이블의 날짜별 행 수와 합계 조회

컨테이너는 두 개예요. PostgreSQL 하나에 Airflow 메타데이터용 airflow DB와 집계 대상인 warehouse DB를 함께 두고, Airflow는 airflow standalone 명령으로 스케줄러, API 서버, DAG 프로세서, triggerer를 한 컨테이너에서 실행해요. standalone은 개발용이에요.

AIRFLOW_CONN_WAREHOUSE 환경 변수는 warehouse라는 이름의 Airflow connection을 만들어요. DAG는 이 connection으로 warehouse DB에 접속합니다.

init.sql은 두 DB를 만들고, warehouse에 원본 주문 테이블과 결과 테이블 네 개를 만들어요. 주문은 2026년 9월 1일부터 5일까지 하루 1,000건씩이고, setseed로 난수 시드를 고정해서 금액이 매번 같게 나와요.

결과 테이블은 아래 DAG마다 하나씩이에요. day 기본 키는 UPSERT용 테이블 daily_sales_up에만 있고, 나머지는 키가 없어요.

dags/sales.py는 쓰기 방식만 다른 DAG를 만들어요.

• sales_append: daily_sales에 INSERT만 해요.
• sales_delete_insert: daily_sales_di에서 그 날짜 행을 지운 뒤 다시 넣어요. 두 문장은 한 트랜잭션이에요.
• sales_upsert: daily_sales_up에 INSERT ... ON CONFLICT (day) DO UPDATE로 써서 같은 날짜 행을 덮어써요. SET에는 충돌해서 넣지 못한 새 행의 값(EXCLUDED)을 그대로 넣습니다.

같은 파일의 sales_now(처리할 날짜를 실행 시간에서 구하는 DAG)와 결과 테이블 daily_sales_now는 이 글에서는 쓰지 않아요.

모두 run(DAG를 한 번 실행한 것)이 처리할 구간(data interval) 안에 들어오는 주문을 세고 금액을 더해 결과 테이블에 써요. schedule은 하루 구간을 만드는 CronDataIntervalTimetable이에요.

fail_after_write라는 Airflow Variable이 true이면, 첫 시도에서 쓰기를 커밋한 다음 일부러 예외를 던져요. 재시도는 1번, 간격은 5초로 두었습니다.

DB 접근에는 이미지에 들어 있는 apache-airflow-providers-postgres 7.0.2의 PostgresHook을 써요. get_conn()이 돌려주는 psycopg2 연결은 with conn: 블록이 정상으로 끝나면 커밋하고, 예외가 나면 롤백합니다.

이제 명령을 짧게 쓸 수 있는 셸 함수를 만들어 놓을게요. COMPOSE_PROJECT_NAME으로 compose 프로젝트 이름을 airf로 고정해 두면, 이후 docker compose 명령은 모두 이 프로젝트를 가리켜요. 아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.

export COMPOSE_PROJECT_NAME=airf

af() { docker compose exec -T airflow airflow "$@"; }

sql() { docker compose exec -T postgres psql -U postgres -d warehouse "$@"; }

report() { sql < report.sql; }

tlog() {
  docker compose exec -T airflow \
    cat "/opt/airflow/logs/dag_id=$1/run_id=$2/task_id=$3/attempt=${4:-1}.log" \
    | grep -E '"logger":"task.stdout"|"level":"error"' \
    | sed -E -e 's/^.*"exc_type":"([^"]*)","exc_value":"([^"]*)".*$/\1: \2/' \
             -e 's/^.*"event":"([^"]*)".*$/\1/'
}

attempts() { docker compose exec -T airflow ls "/opt/airflow/logs/dag_id=$1/run_id=$2/task_id=$3"; }

waitruns() {
  until [ "$(docker compose exec -T postgres psql -U postgres -d airflow -tAc \
    "SELECT (SELECT count(*) FROM dag_run WHERE state IN ('queued', 'running'))
          + (SELECT count(*) FROM backfill WHERE completed_at IS NULL)")" = 0 ]; do sleep 2; done
}

Airflow 3의 태스크 로그는 한 줄에 JSON 하나씩 기록돼요. 그래서 tlog <dag_id> <run_id> <task_id> [시도 번호]는 그중 태스크가 print로 남긴 줄과 예외 줄만 골라 보여 줘요. report.sql은 세 결과 테이블의 날짜별 행 수와 합계를 한 번에 봅니다. attempts <dag_id> <run_id> <task_id>는 시도마다 생긴 로그 파일 목록을 보여 주고, waitruns는 대기·실행 중인 run과 끝나지 않은 backfill이 없을 때까지 기다려요.

컨테이너를 띄우고 API 서버가 응답할 때까지 기다릴게요. 처음에는 메타데이터 DB를 만드는 과정이 있어서 1분쯤 걸려요.

docker compose up -d
until curl -sf localhost:7260/api/v2/monitor/health > /dev/null; do sleep 3; done

원본의 날짜별 건수와 합계를 볼게요. 이 값이 정답이에요.

sql -c "SELECT ordered_at::date AS day, count(*) AS orders, sum(amount) AS amount FROM orders_raw GROUP BY 1 ORDER BY 1;"

    day     | orders |  amount  
------------+--------+----------
 2026-09-01 |   1000 | 25914100
 2026-09-02 |   1000 | 25784300
 2026-09-03 |   1000 | 25281800
 2026-09-04 |   1000 | 25153000
 2026-09-05 |   1000 | 25948100
(5 rows)

새 DAG는 일시 정지 상태로 등록되기 때문에([core] dags_are_paused_at_creation 기본값 True) 실습할 DAG를 unpause해 주세요. 실습한 날은 2026년 10월 2일이에요.

unpause하면 세 DAG가 10월 1일 run을 하나씩 실행해요. catchup=False라서 지난 구간을 모두 채우지 않고, 가장 최근 구간 하나만 실행하는 거예요. 원본에 10월 데이터가 없으니 0건 행이 생기므로, 비교하기 쉽도록 결과 테이블을 비우고 시작할게요.

for d in sales_append sales_delete_insert sales_upsert; do af dags unpause $d > /dev/null; done
sleep 15; waitruns
sql -c "TRUNCATE daily_sales, daily_sales_di, daily_sales_up;"

실습 1: INSERT만 하는 태스크를 재시도하면

fail_after_write를 켜고 세 DAG의 9월 1일을 backfill해 볼게요. 첫 시도는 쓰기를 커밋한 뒤 실패하고, 5초 뒤 재시도가 실행돼요.

af variables set fail_after_write true
for d in sales_append sales_delete_insert sales_upsert; do
  af backfill create --dag-id $d --from-date 2026-09-01 --to-date 2026-09-01 > /dev/null
done
sleep 5; waitruns
attempts sales_append backfill__2026-09-02T00:00:00+00:00 aggregate

attempt=1.log
attempt=2.log

시도마다 로그 파일이 따로 생겼어요. run ID가 backfill__2026-09-02...인 건 run ID가 구간의 끝으로 붙기 때문이에요. 두 시도의 로그를 볼게요.

tlog sales_append backfill__2026-09-02T00:00:00+00:00 aggregate 1
tlog sales_append backfill__2026-09-02T00:00:00+00:00 aggregate 2

logical_date=2026-09-01 00:00:00+00:00 data_interval=[2026-09-01 00:00:00+00:00, 2026-09-02 00:00:00+00:00) now=2026-10-02 01:31:34.099455+00:00 try=1
wrote day=2026-09-01
RuntimeError: failed after commit
logical_date=2026-09-01 00:00:00+00:00 data_interval=[2026-09-01 00:00:00+00:00, 2026-09-02 00:00:00+00:00) now=2026-10-02 01:31:39.765825+00:00 try=2
wrote day=2026-09-01

두 시도 모두 9월 1일을 썼어요. Airflow 입장에서 첫 시도는 실패했으니, run은 재시도가 성공한 시점에 한 번 성공한 것으로 끝나요. 결과 테이블을 볼게요.

report

      dag      |    day     | rows | orders |  amount  
---------------+------------+------+--------+----------
 append        | 2026-09-01 |    2 |   2000 | 51828200
 delete_insert | 2026-09-01 |    1 |   1000 | 25914100
 upsert        | 2026-09-01 |    1 |   1000 | 25914100
(3 rows)

출력을 보면 sales_append는 9월 1일 행이 2개이고, 합계 51,828,200원은 정답 25,914,100원의 정확히 두 배예요. 같은 조건에서 sales_delete_insert와 sales_upsert는 1행이고 정답과 같아요.

"실패한 태스크만 다시 도니까 재시도는 안전하다"는 말은, 실패한 시도가 아무것도 남기지 않았을 때만 맞아요.

태스크는 DB에 쓰기를 커밋한 다음에도 실패할 수 있어요. 커밋 다음 줄에서 난 예외나, 커밋 직후 워커 프로세스가 죽는 경우가 그래요. Airflow는 태스크가 어디까지 했는지 모르고, 처음부터 다시 실행할 뿐이에요.

이 상황은 Kafka 컨슈머가 처리를 마치고 offset을 커밋하기 전에 죽는 경우와 같은 구조예요. 컨슈머 쪽 설계가 더 궁금하시면 at-least-once 처리를 다룬 글을 참고해 주세요.

다음 실습을 위해 실패 스위치를 꺼 둘게요.

af variables set fail_after_write false

실습 2: 성공한 run을 clear해도 같은 일이 생깁니다

원본 데이터를 고쳤거나 집계 코드를 바꿨을 때, 이미 성공한 run의 태스크를 clear해서 다시 실행하는 일이 흔해요. airflow tasks clear의 -s, -e는 logical date 범위예요.

for d in sales_append sales_delete_insert sales_upsert; do
  af tasks clear -s 2026-09-01 -e 2026-09-01 -y $d
done
sleep 5; waitruns
report

      dag      |    day     | rows | orders |  amount  
---------------+------------+------+--------+----------
 append        | 2026-09-01 |    3 |   3000 | 77742300
 delete_insert | 2026-09-01 |    1 |   1000 | 25914100
 upsert        | 2026-09-01 |    1 |   1000 | 25914100
(3 rows)

sales_append는 3행으로 늘었고, 멱등 버전 두 개는 그대로예요. clear는 실패와 상관없이 사람이 요청하는 재실행이고, INSERT만 하는 태스크는 clear할 때마다 행이 늘어나요.

실습 3: backfill로 같은 구간을 다시 처리하면

세 DAG를 9월 1일부터 5일까지 backfill하고, 같은 구간을 --reprocess-behavior completed로 한 번 더 backfill해 볼게요. 이 옵션의 값은 세 가지예요.

• none(기본값): run이 이미 있는 날짜는 건너뛰어요.
• failed: run이 없는 날짜와 실패한 run을 다시 실행해요. 성공한 run은 건너뛰어요.
• completed: 성공한 run까지 모두 다시 실행해요.

for d in sales_append sales_delete_insert sales_upsert; do
  af backfill create --dag-id $d --from-date 2026-09-01 --to-date 2026-09-05 > /dev/null
done
sleep 5; waitruns
for d in sales_append sales_delete_insert sales_upsert; do
  af backfill create --dag-id $d --from-date 2026-09-01 --to-date 2026-09-05 --reprocess-behavior completed > /dev/null
done
sleep 5; waitruns
report

      dag      |    day     | rows | orders |  amount   
---------------+------------+------+--------+-----------
 append        | 2026-09-01 |    4 |   4000 | 103656400
 append        | 2026-09-02 |    2 |   2000 |  51568600
 append        | 2026-09-03 |    2 |   2000 |  50563600
 append        | 2026-09-04 |    2 |   2000 |  50306000
 append        | 2026-09-05 |    2 |   2000 |  51896200
 delete_insert | 2026-09-01 |    1 |   1000 |  25914100
 delete_insert | 2026-09-02 |    1 |   1000 |  25784300
 delete_insert | 2026-09-03 |    1 |   1000 |  25281800
 delete_insert | 2026-09-04 |    1 |   1000 |  25153000
 delete_insert | 2026-09-05 |    1 |   1000 |  25948100
 upsert        | 2026-09-01 |    1 |   1000 |  25914100
 upsert        | 2026-09-02 |    1 |   1000 |  25784300
 upsert        | 2026-09-03 |    1 |   1000 |  25281800
 upsert        | 2026-09-04 |    1 |   1000 |  25153000
 upsert        | 2026-09-05 |    1 |   1000 |  25948100
(15 rows)

첫 backfill은 run이 없는 9월 2일부터 5일까지만 새로 만들고, 이미 성공한 9월 1일은 건너뛰었어요. completed를 준 두 번째 backfill은 다섯 날짜를 모두 다시 실행했어요.

그래서 sales_append는 모든 날짜에 행이 하나씩 더 쌓였고, 9월 1일은 재시도, clear, 다시 처리를 거쳐 4행이 됐어요. 멱등 버전 둘은 5행 그대로이고, 날짜별 합계가 원본과 같아요.

정리

범위: Airflow 3.3.2(apache/airflow:3.3.2-python3.12), PostgreSQL 17, 9월 1~5일 하루 1,000건씩의 주문, 쓰기 방식만 다른 DAG 세 개.

• 쓰기를 커밋한 뒤 실패한 첫 시도를 재시도하자 sales_append의 9월 1일은 2행, 합계 51,828,200원으로 정답의 두 배가 됐어요. 재시도를 켜기 전에 태스크의 쓰기부터 멱등하게 만드세요.
• 성공한 run을 clear하자 sales_append는 3행이 됐고, --reprocess-behavior completed backfill까지 거치자 9월 1일이 4행이 됐어요. 다시 실행될 수 있는 태스크에서 결과 테이블에 INSERT만 하지 마세요.
• 구간 DELETE 후 INSERT와 UPSERT는 재시도, clear, 다시 처리를 모두 거쳐도 날짜마다 1행이고 합계가 원본과 같았어요. DELETE와 INSERT는 한 트랜잭션에 묶고, UPSERT의 SET에는 EXCLUDED 값을 그대로 넣으세요.
• 기본 backfill은 이미 성공한 9월 1일을 건너뛰었고 --reprocess-behavior completed는 5개 run을 모두 다시 실행했어요. 이미 집계한 날짜를 다시 계산하려면 completed를 주세요.