Airflow

태스크/에셋 상태 저장소란

태스크/에셋 상태 저장소는 태스크 상태 저장소와 에셋 상태 저장소의 영속화 계층이다. 기본적으로 둘 다 Airflow 메타데이터 DB에 저장된다. 설정 섹션은 ``[state_store]``이고 (``[task_state_store]``가 아니다), CLI는 ``airflow state-store``이며, 백엔드 베이스 클래스는 ``BaseStoreBackend``다 — 이 모두가 "state_store"라는 이름을 공유해서 쓴다.

핵심 설정 옵션

``backend``는 ``BaseStoreBackend``를 구현한 클래스로의 전체 dotted 경로이며, 기본값은 내장 metastore 백엔드다.

``default_retention_days``는 태스크 상태 저장소 행이 만료되기까지의 일수다. 명시적 retention 없이 쓰인 키는 워커에서 ``now() + default_retention_days``로 ``expires_at``이 계산된다. 0으로 설정하면 시간 기반 정리를 완전히 끈다. 기본값은 30이며, 이 설정은 에셋 상태 저장소 행에는 적용되지 않는다.

``clear_on_success``가 True면 태스크 인스턴스가 success 상태가 될 때 그 태스크 인스턴스의 모든 태스크 상태 저장소 키가 자동으로 삭제된다. 기본값은 False로, 실행이 끝난 뒤에도 관측가능성을 위해(예: 제출한 job ID나 마지막 행 수를 실행 완료 후에도 UI/REST API에서 읽을 수 있도록) 항목을 남겨둔다. 이 옵션은 태스크 상태 저장소만 지운다 — 에셋 상태 저장소에는 영향을 주지 않는다. 에셋 저장소는 태스크 인스턴스가 아니라 에셋에 스코프되어 있어 명시적으로 지워야 한다.

``state_cleanup_batch_size``는 가비지 컬렉션 정리 시 배치당 삭제되는 행 수다. 기본값 0은 매칭되는 모든 행을 단일 statement로 삭제한다는 뜻이며, ``task_state_store`` 테이블이 큰 배포에서는 락 경합을 줄이기 위해 이 값을 조정할 수 있다.

워커사이드 백엔드 ([workers] state_backend)

``[workers]`` 아래의 별도 선택적 설정 키로, 태스크 상태 저장소와 에셋 상태 저장소 값이 API 서버에 도달하기 전에 워커사이드 백엔드를 거치도록 라우팅할 수 있다.

이 설정이 있으면, ``TaskStateStoreAccessor.set()``은 값을 Execution API로 보내기 전에 워커사이드 백엔드의 ``serialize_task_state_store_to_ref()``를 호출해 반환된 참조값(실제 저장소를 가리키는 레퍼런스)을 대신 보내고, ``get()``은 Execution API로부터 저장된 참조를 받은 뒤 ``deserialize_task_state_store_from_ref()``를 호출한다.

가비지 컬렉션 규칙

정리("가비지 컬렉션") 작업은 Airflow CLI로 트리거되며 ``airflow state-store clean`` 명령을 사용한다. 이 과정은 다음 규칙에 따라 저장소 행을 제거한다.

시간 기반 만료(태스크 상태 저장소만 해당): ``expires_at < now()``인 행이 삭제된다. ``expires_at``은 서버가 아니라 **워커**에서 쓰기 시점에 계산된다. ``default_retention_days`` 폴백(태스크 상태 저장소만 해당): 명시적 retention 없이 쓰인 키는 쓰기 시점에 ``now() + default_retention_days``의 ``expires_at``을 받는다. ``NEVER_EXPIRE`` 키: ``retention=NEVER_EXPIRE``로 설정된 키는 ``expires_at = NULL``과 함께 가비지 컬렉션이 무조건 건너뛰도록 하는 플래그를 갖는다. ``default_retention_days``와 무관하게 시간 기반 정리로 절대 삭제되지 않는다. ``on_delete=CASCADE``(에셋 상태 저장소): 에셋이 삭제되면 그 에셋에 해당하는 모든 에셋 상태 저장소 행이 함께 삭제된다.

중요: 가비지 컬렉션은 오직 ``MetastoreBackend``에서만 동작한다. 커스텀 백엔드는 명시적으로 건너뛰어진다.

커스텀 백엔드 구현

커스텀 백엔드는 ``BaseStoreBackend``를 상속하고 ``get``, ``set``, ``delete``, ``clear``(동기)와 ``aget``, ``aset``, ``adelete``, ``aclear``(비동기) 추상 메서드를 구현해야 한다. 각 메서드는 ``TaskScope`` 또는 ``AssetScope``인 ``scope`` 인자를 받으며, ``isinstance``로 분기해 처리한다.

``AssetScope``는 3개의 선택적 필드를 갖는다: ``asset_id``(정수, 서버사이드 전용), ``name``, ``uri``. 이 중 최소 하나는 설정되어야 한다. 서버사이드 연산(REST API 호출)은 ``asset_id``를 제공하고, 워커사이드 연산은 ``name``이나 ``uri``를 제공한다(워커는 정수 ``asset_id``에 접근할 수 없기 때문). 클래스는 ``[state_store] backend``로 설정한다.

커스텀 워커사이드 백엔드

워커사이드 백엔드는 ``BaseStoreBackend``를 확장해 두 쌍의 직렬화 훅을 추가로 구현한다. 이들은 ``[workers] state_backend``로 별도로 설정되며 API 서버가 아니라 **워커 프로세스에서** 실행된다. 이를 통해 대용량 페이로드나 자격증명이 포함된 데이터를 워커 인프라를 이용해 직접 저장하면서, DB에는 컴팩트한 참조 문자열만 남길 수 있다.

오버라이드할 4개의 직렬화 훅: ``serialize_task_state_store_to_ref``(``TaskStateStoreAccessor.set()``이 값을 Execution API로 보내기 전에 호출; 컴팩트한 참조 문자열, 예를 들어 S3 키를 반환해 DB에 실제 값 대신 저장), ``deserialize_task_state_store_from_ref``(``TaskStateStoreAccessor.get()``이 백엔드로부터 참조를 받은 뒤 호출; 실제 값을 반환), ``serialize_asset_state_store_to_ref``(태스크용과 동일하지만 에셋 상태 저장소용이며, scope로 name/uri를 가진 AssetScope를 받음), ``deserialize_asset_state_store_from_ref``(``AssetStateStoreAccessor.get()``이 저장된 참조를 실제 값으로 되돌릴 때 호출).

중요: **참조는 결정적(deterministic)이어야 한다.** 같은 입력(scope + key)이 주어지면 직렬화 메서드는 항상 같은 참조 문자열을 반환해야 한다. 타임스탬프, 랜덤 UUID 등 비결정적 요소를 참조 경로에 넣으면 안 된다. 키가 삭제되거나 클리어될 때 Airflow는 먼저 DB의 참조를 지운 다음 백엔드의 ``delete()``나 ``clear()``를 호출한다. DB 행이 사라진 뒤 백엔드 정리가 실패하면 외부 객체가 고아가 되는데, 참조가 결정적이라면 같은 키에 대한 다음 ``set()``이 그 고아 객체를 덮어써 복구 가능해진다. 비결정적 참조였다면 그 외부 객체는 찾을 방법 없이 영구히 고아로 남는다.