Airflow 3.3 State Store: 재시도와 실행을 넘어 상태를 어디에 둘 것인가

새 저장소가 필요한 순간은 실패 뒤에 드러난다

2026년 7월 6일 공개된 Apache Airflow 3.3.0은 AIP-103에 따라 Task State Store와 Asset State Store를 추가했다. 이름만 보면 XCom이나 Variable의 또 다른 변형처럼 보이지만, 이 기능이 메우는 구멍은 다르다.

외부 Spark job을 제출한 task가 worker 재시작으로 죽었다고 가정해 보자. 재시도된 task가 이전 job ID를 잃었다면 같은 job을 또 제출할 수 있다. 매일 증분 적재하는 task가 마지막 watermark를 task return value에만 남겼다면, 재시도나 수동 재실행에서 어느 지점부터 읽어야 하는지가 흐려진다. 그동안 이런 상태는 XCom, Variable, 별도 Redis·DB 테이블에 제각각 저장됐다.

Airflow 3.3의 State Store는 이 상태를 두 가지 소유권으로 나눈다.

Task State Store: 하나의 task instance가 재시도와 worker crash를 넘어 이어서 일하기 위한 실행 상태
Asset State Store: 특정 DagRun이 아니라 asset 자체가 다음 실행에 넘겨주는 장기 상태

중요한 변화는 "Airflow task도 이제 stateful하게 작성하라"가 아니다. 오히려 상태의 수명과 소유자를 명시하고, XCom·Variable과 섞지 말라는 경계가 생긴 것이다. 이 장에서는 새 저장소의 구조, 실패 경계, 메타데이터 DB 영향, 마이그레이션과 운영 체크리스트를 살펴본다.

이 장은 Airflow 3.3.0 문서와 소스를 기준으로 한다. 3.3.0은 공개된 지 8일 된 릴리스이므로, 도입할 때는 사용 중인 patch release의 release note와 provider constraints를 다시 확인해야 한다.

먼저 XCom·Variable과 역할을 분리한다

State Store를 도입할 때 첫 질문은 API 사용법이 아니라 이 값의 소유자와 수명은 누구인가다.

메커니즘	소유 범위	적합한 값	실패·재실행 관점
XCom	task 간 데이터 전달	한 task의 출력, downstream 입력	공식 3.3 문서는 retry를 넘는 영속 상태에 쓰지 말라고 명시한다
Variable	Airflow 설치 전체	운영자가 드물게 바꾸는 설정	task가 실행 cursor를 계속 덮어쓰는 용도가 아니다
Task State Store	`dag_id + run_id + task_id + map_index`	외부 job ID, 실행 중 checkpoint, 진행률	같은 task instance의 retry와 worker crash를 넘어 유지된다
Asset State Store	asset	watermark, 증분 cursor, asset별 메타데이터	DagRun과 독립적으로 다음 실행까지 유지된다

XCom으로 잘 동작하는 task 간 전달을 전부 옮길 이유는 없다. Airflow 문서도 기존 XCom 패턴이 문제없이 작동한다면 Task State Store로 마이그레이션할 필요가 없다고 설명한다.

반대로 다음 질문에 "예"라면 State Store 후보가 된다.

task가 실패한 뒤 같은 외부 작업에 다시 연결해야 하는가?
retry가 처음부터 다시 실행되지 않고 중간 checkpoint부터 재개되어야 하는가?
cursor가 특정 task run이 아니라 논리적인 데이터 asset에 속하는가?
실행 중 진행률을 UI나 API에서 보고 싶지만, downstream 전달값은 아닌가?

이 구분을 하지 않으면 State Store가 기존 임시 상태를 한곳에 모으는 쓰레기통이 된다. 저장할 수 있다는 사실과 저장해야 한다는 판단은 다르다.

구조: task 경계와 asset 경계를 별도 key space로 만든다

기본 구성에서 worker의 Task SDK accessor는 Execution API를 통해 값을 읽고 쓰고, API server 쪽 State Store backend가 이를 Airflow metadata database에 보관한다. UI와 Core API도 같은 상태를 조회·관리한다.

Airflow 3.3 Task·Asset State Store의 저장 경로와 수명 경계

3.3.0 migration source를 보면 두 범위가 실제 테이블 구조에도 반영되어 있다.

task_state_store: dag_run_id, task_id, map_index, key 조합을 유일하게 관리하고 expires_at index를 둔다.
asset_state_store: asset_id + key가 primary key이며, asset 삭제 시 ON DELETE CASCADE로 정리된다.
두 테이블의 value는 JSON을 직렬화한 text다. MySQL에서는 MEDIUMTEXT variant를 사용한다.

따라서 State Store 사용량은 애플리케이션 코드의 작은 편의 기능이 아니라 metadata DB의 row 수, write rate, vacuum·purge 부하로 돌아온다. 진행률을 매 row마다 기록하거나 큰 payload를 반복 저장하면 scheduler와 UI가 함께 의존하는 DB에 새로운 쓰기 workload를 추가한다.

Task State Store: retry가 같은 외부 작업에 다시 붙게 한다

Task State Store는 task context의 context["task_state_store"]로 접근하며 get, set, delete, clear를 제공한다. 값은 None을 제외한 JSON-compatible type이어야 한다.

외부 job 재연결 패턴은 다음처럼 작성할 수 있다.

from airflow.sdk import NEVER_EXPIRE, task

@task(retries=3)
def run_external_job(**context):
    store = context["task_state_store"]
    job_id = store.get("job_id")

    if job_id is None:
        run_id = context["dag_run"].run_id
        job_id = external_client.submit(
            idempotency_key=f"daily-load:{run_id}"
        )
        store.set("job_id", job_id, retention=NEVER_EXPIRE)

    result = external_client.wait(job_id)
    store.set("progress", {"state": result.state})
    return result.summary

retry된 task는 기존 job_id를 읽어 새 job을 만들지 않고 이전 job을 기다린다. 이 패턴은 worker process가 죽어도 Airflow 쪽 상태가 남는다는 점에서 XCom보다 목적에 맞다.

하지만 State Store가 exactly-once submit을 보장하지는 않는다. external_client.submit()은 성공했지만 store.set() 전에 worker가 죽는 실패 창이 남는다. retry는 job_id를 찾지 못하고 다시 submit할 수 있다. 따라서 외부 시스템이 지원하는 idempotency key, deterministic job name, submit 결과 조회 API 중 하나와 결합해야 한다.

checkpoint도 같은 원칙을 따른다.

@task(retries=2)
def ingest_pages(**context):
    store = context["task_state_store"]
    last_page = store.get("last_page", default=0)

    for page in range(last_page + 1, total_pages + 1):
        rows = fetch_page(page)
        commit_idempotently(rows, page=page)
        store.set("last_page", page)

여기서는 sink commit이 끝난 뒤에 checkpoint를 전진시킨다. 반대 순서면 checkpoint만 앞서가고 실제 데이터가 빠질 수 있다. 그렇더라도 sink commit과 State Store write는 하나의 transaction이 아니므로, commit 뒤 state write 전에 죽으면 같은 page를 다시 처리한다. commit_idempotently가 필요한 이유다.

retention은 "얼마나 오래 디버깅할 것인가"까지 포함한다

set()의 retention은 세 가지 방식으로 결정된다.

설정	동작
`retention=timedelta(...)`	write 시점부터 지정 기간 뒤 만료
`retention=NEVER_EXPIRE`	`expires_at`을 비워 시간 기반 cleanup에서 제외
`retention=None`	`[state_store] default_retention_days` 사용, 기본 30일

retention=7처럼 정수를 넘기면 안 된다. 공식 API는 timedelta만 받으며 정수는 TypeError를 낸다.

[state_store] clear_on_success = True를 켜면 task가 success로 전환될 때 해당 task instance의 key를 모두 지운다. 저장 공간은 줄지만 성공 뒤 job ID·row count를 UI와 REST API에서 조사할 수 없게 된다. 장애 조사 기간과 metadata DB 비용을 함께 보고 결정해야 한다. 기본값은 False다.

동적 mapped task는 map_index마다 별도 namespace를 가진다. task 내부의 clear()는 현재 index만 지운다. 수천 개 mapped task가 각각 진행률을 자주 기록한다면 key 수는 예상보다 빠르게 커질 수 있다.

Asset State Store: cursor의 소유자를 DAG가 아니라 데이터로 바꾼다

Asset State Store는 DagRun과 독립적으로 asset에 귀속된다. 구체적인 Asset을 task의 inlets나 outlets로 선언한 뒤 context["asset_state_store"][asset]로 접근한다.

from airflow.sdk import Asset, DAG, task

orders = Asset("orders", uri="s3://data/orders/")

with DAG("incremental_orders", schedule="@daily", max_active_runs=1):
    @task(inlets=[orders], outlets=[orders])
    def load_orders(**context):
        store = context["asset_state_store"][orders]
        watermark = store.get(
            "watermark",
            default="1970-01-01T00:00:00Z",
        )

        rows = fetch_orders_after(watermark)
        if not rows:
            return

        merge_idempotently(rows)
        store.set("watermark", max(row["created_at"] for row in rows))

inlet 또는 outlet이 하나뿐이면 context["asset_state_store"]에서 바로 get과 set을 호출하는 shorthand도 있다. 그러나 asset이 여러 개인 task에서 shorthand를 쓰면 ValueError가 나므로, 운영 코드에서는 asset을 명시적으로 subscripting하는 편이 리뷰하기 쉽다. inlet/outlet 자체가 없으면 accessor를 사용할 수 없고 runtime에 KeyError가 발생한다.

Asset State Store에는 task state의 retention parameter가 없다. 값은 명시적으로 delete·clear하거나 asset이 비활성화되어 orphan sweep 대상이 될 때까지 유지된다. [state_store] clear_on_success도 asset state에는 영향을 주지 않는다.

이 긴 수명은 편리하지만 동시성 위험도 만든다. 두 DagRun이 같은 asset watermark를 동시에 읽고 서로 다른 값으로 덮어쓰면 마지막 write가 이전 진행을 되돌릴 수 있다. 단순 key/value API와 외부 sink commit 사이에는 transaction 경계가 없다.

그래서 cursor를 asset store에 둘 때는 다음 중 적어도 하나가 필요하다.

max_active_runs=1처럼 동일 cursor의 writer를 직렬화한다.
cursor가 단조 증가하도록 write 전 현재 값을 다시 확인한다.
source offset과 sink commit을 외부 transaction·checkpoint ledger에서 원자적으로 관리하고, asset store는 관측용 복제본으로만 쓴다.
날짜 문자열보다 source가 정의한 offset, LSN, partition ID처럼 비교 규칙이 명확한 값을 사용한다.

Airflow 3.3은 watcher trigger의 run() 안에서도 asset state를 읽고 쓸 수 있게 했다. polling source의 last_seen_id를 asset에 저장하면 triggerer 재시작 뒤 이어서 볼 수 있다. 단, accessor는 BaseEventTrigger.run() 안에서만 주입되며 __init__, serialize(), 일반 BaseTrigger에는 제공되지 않는다.

큰 값과 credentialed data를 metadata DB에 그대로 넣지 않는다

기본 backend는 JSON 값을 metadata DB에 저장한다. migration schema의 value는 text이고 문서는 이 값이 암호화된다고 약속하지 않는다. 따라서 password, access token, connection secret을 State Store 값으로 넣지 않는 것이 안전하다. UI와 Core API에서 누가 읽을 수 있는지도 권한 검토 대상이다.

Airflow 3.3에는 저장 경로를 바꾸는 두 설정이 있다.

설정	실행 위치	목적
`[state_store] backend`	API server 쪽	기본 `MetastoreBackend` 대신 전체 store backend 교체
`[workers] state_backend`	worker 쪽	실제 payload를 외부 저장소에 두고 compact reference만 API에 전달

worker-side backend는 serialize_task_state_store_to_ref()와 asset 대응 method로 값을 외부 저장소에 쓰고 deterministic reference를 반환한다. read 때는 reference를 실제 값으로 복원한다.

reference가 deterministic해야 하는 이유는 delete 순서 때문이다. Airflow는 DB reference를 먼저 지우고 외부 backend의 delete()나 clear()를 호출한다. 외부 삭제가 실패하면 orphan object가 남는다. 같은 scope + key가 항상 같은 path를 만들면 다음 set()이 그 object를 덮어써 복구할 수 있지만, timestamp나 random UUID가 들어가면 영구 orphan이 된다.

외부 backend가 만능은 아니다.

API server와 worker의 장애 경계가 외부 저장소까지 넓어진다.
DB reference는 사라졌지만 object는 남는 cleanup failure를 감시해야 한다.
직렬화 format을 바꾸면 과거 reference를 읽는 compatibility code가 필요하다.
State Store를 blob store처럼 쓰기보다 checkpoint ID나 작은 cursor를 저장하는 편이 단순하다.

cleanup은 자동 스케줄이 아니다

Task State Store에 expires_at이 있다고 해서 row가 저절로 사라지는 것은 아니다. Airflow 3.3 문서는 cleanup 실행을 운영자 책임으로 둔다.

# 삭제 대상만 확인
airflow state-store clean --dry-run

# 만료된 task state row 삭제
airflow state-store clean

기본 state_cleanup_batch_size = 0은 대상 row를 한 statement로 지운다. 큰 테이블에서는 transaction과 lock 유지 시간이 길어질 수 있으므로 batch size를 명시한다.

[state_store]
default_retention_days = 30
clear_on_success = False
state_cleanup_batch_size = 10000

cleanup 명령은 기본 MetastoreBackend의 만료된 task state만 대상으로 한다. asset state는 이 명령으로 시간 만료되지 않는다. custom [state_store] backend도 기본 cleanup에서 건너뛰므로 자체 maintenance process가 필요하다.

운영 지표는 최소한 다음을 포함해야 한다.

task_state_store row 수와 일별 증가량
만료되었지만 아직 삭제되지 않은 row 수
cleanup 실행 시간, batch 수, 실패 횟수
metadata DB write latency와 lock wait 변화
key별 payload 크기와 write 빈도
asset 비활성화 뒤 남은 orphan state
worker-side backend object 수와 DB reference 수의 차이

특히 progress를 짧은 interval로 갱신하면 같은 key를 overwrite하더라도 write amplification이 생긴다. 사람이나 자동화가 실제로 소비하는 해상도에 맞춰 throttle해야 한다.

3.3 업그레이드와 도입 순서

State Store를 쓸 계획이 없어도 3.3.0 schema에는 task_state_store와 asset_state_store table이 추가된다. 새 binary와 scheduler를 올리기 전에 일반적인 Airflow DB migration 절차와 rollback backup을 준비해야 한다.

1단계: 기능 없이 업그레이드 경계를 검증한다

production metadata DB snapshot 또는 검증된 backup을 만든다.
staging clone에서 Airflow 3.3 migration을 실행하고 소요 시간과 lock을 측정한다.
scheduler, API server, triggerer, worker와 Task SDK image가 같은 release constraints를 따르는지 확인한다.
기존 DAG parse, task 실행, XCom, remote logging, provider compatibility를 먼저 회귀 시험한다.
State Store를 사용하지 않은 상태로 canary를 안정화한다.

State Store는 새 기능이므로 기존 XCom을 한 번에 치환하는 migration이 필요하지 않다. 먼저 기능 도입과 platform version upgrade를 분리하면 문제가 schema·runtime 때문인지 DAG code 때문인지 좁히기 쉽다.

2단계: 하나의 retry 복구 사례로 시작한다

외부 job ID처럼 실패 비용이 명확한 task 하나를 고른다.

현재 중복 제출이 가능한 실패 창을 문서화한다.
idempotency key와 외부 job 조회 경로를 먼저 만든다.
job_id, submitted_at, progress처럼 key schema를 정한다.
retention과 성공 후 조사 기간을 정한다.
worker kill 뒤 retry가 같은 job에 attach하는지 fault injection으로 확인한다.

3단계: asset cursor는 별도 검증한다

watermark는 잘못 움직이면 중복보다 데이터 누락이 더 위험하다.

sink commit 전에는 watermark를 전진시키지 않는다.
concurrent DagRun과 manual rerun의 writer 규칙을 정한다.
watermark가 뒤로 이동했을 때 alert한다.
source와 sink를 reconciliation해 누락·중복을 검출한다.
cursor 초기화와 강제 rewind는 UI click이 아니라 승인된 runbook으로 수행한다.

4단계: cleanup을 배포와 동시에 운영한다

첫 State Store DAG를 배포한 날부터 --dry-run, 정기 cleanup, table growth dashboard를 준비한다. retention만 설정하고 cleanup schedule을 빼먹으면 30일 뒤에도 row는 남아 있다.

실패 주입으로 확인할 검증 행렬

happy path 한 번으로는 State Store의 목적을 검증할 수 없다. 실패가 발생하는 정확한 위치를 바꿔가며 확인해야 한다.

주입 지점	기대 동작	반드시 확인할 부작용
외부 job submit 직후, state write 전 worker kill	retry가 idempotency key로 기존 job을 찾는다	duplicate job이 생기지 않았는가
job ID 저장 뒤 worker kill	retry가 같은 job에 attach한다	새 job submit 호출이 없었는가
sink commit 뒤 checkpoint write 전 kill	같은 batch를 재처리한다	idempotent merge로 중복이 없는가
checkpoint write 뒤 다음 batch 전 kill	저장된 다음 지점부터 재개한다	page·offset 누락이 없는가
asset writer 두 개 동시 실행	직렬화되거나 단조 증가 규칙이 지켜진다	watermark가 뒤로 이동하지 않았는가
cleanup 중 metadata DB 부하 증가	batch 단위로 진행하고 scheduler SLO를 지킨다	lock wait·replica lag가 임계값을 넘지 않는가
worker-side external delete 실패	orphan을 탐지하고 같은 ref로 복구한다	DB와 object count가 계속 벌어지지 않는가

rollback도 DAG code만 이전으로 되돌리는 것으로 끝나지 않는다. 이미 생성된 state row와 외부 reference를 누가, 언제, 어떤 기준으로 지울지 정해야 한다. rollback version이 새 key format을 모르면 값을 무시할 수는 있지만, 정리 책임까지 사라지는 것은 아니다.

도입 체크리스트

경계 정의

task 간 전달은 XCom, 설치 설정은 Variable, retry 상태는 Task Store, cross-run cursor는 Asset Store로 분리한다.

각 key의 owner, JSON schema, 수명, 삭제 책임자를 문서화한다.

정확성

외부 submit에는 idempotency key를 넣고 checkpoint는 sink commit 뒤에 갱신한다.

concurrent asset writer를 직렬화하거나 단조 증가·reconciliation 규칙을 둔다.

worker kill과 retry를 포함한 실패 주입 시험을 통과한다.

용량·보안

retention, clear_on_success, cleanup batch와 실행 주기를 함께 정한다.

secret과 큰 payload를 metadata DB에 넣지 않고 UI·API read 권한을 확인한다.

custom backend의 deterministic reference와 orphan 탐지 절차를 검증한다.

변경·롤백

DB migration과 DAG 기능 도입을 분리하고 기존 DAG 회귀 시험을 먼저 통과한다.

rollback 뒤 남는 state row, external object, cursor 해석과 삭제 절차를 runbook에 넣는다.

Airflow 3.3 State Store 운영 gate

정리

Airflow 3.3 State Store의 가치는 key/value API 네 개에 있지 않다. 외부 job ID와 checkpoint를 task instance의 실행 상태로, watermark와 watcher cursor를 asset의 장기 상태로 구분한 데 있다. XCom이나 Variable에 섞여 있던 값의 수명과 책임을 더 명확하게 만들 수 있다.

그러나 저장소가 생겼다고 분산 시스템의 원자성이 생기는 것은 아니다. 외부 submit과 state write, sink commit과 watermark write 사이에는 여전히 실패 창이 있다. idempotency, 단조 cursor, writer 직렬화, reconciliation이 함께 있어야 한다.

운영자가 기억할 것은 세 가지다. 첫째, working XCom을 무조건 옮기지 않는다. 둘째, retention 설정만으로 cleanup이 실행된다고 가정하지 않는다. 셋째, State Store를 metadata DB의 공짜 공간으로 보지 않는다. 기능 도입의 성공 기준은 값이 저장되는지가 아니라 실패 뒤 같은 작업으로 안전하게 복귀하고, 누락 없이 정리되는지다.