재개 가능한 태스크

재개 가능한 태스크 (Resumable Tasks)

데이터 엔지니어링 워크플로를 보면, 실제 작업은 외부 시스템이 하고 Airflow는 결과를 기다리는 형태가 정말 많아요. Spark 잡, BigQuery 쿼리, Kubernetes 배치 파드, EMR 스텝 같은 게 대표적이죠. 이런 태스크들은 공통된 실패 형태를 갖고 있어요. 워커 프로세스가 재시작되거나 호스트가 선점되면 처음부터 다시 시도하면서 지금까지의 진행을 통째로 잃어버린다는 거예요. 외부 시스템 입장에서는 중복 실행이 생길 수도 있고요. Airflow 3.3에서는 이런 긴 외부 작업을 다루는 세 가지 접근법을 제안해요.

출처: Resumable Tasks - Apache Airflow 공식 문서

Airflow 3.3.0에서 추가.

지연 가능한 연산자 (Deferrable Operators)

지연 가능한 연산자(Deferrable Operator)는 폴링을 시작할 지점에서 스스로 멈추고, 폴링 작업을 Triggerer 컴포넌트에 넘긴 뒤 워커 슬롯을 놓아줘요. 외부 조건이 충족되면 Triggerer가 태스크를 깨우고, 워커는 연산자가 멈췄던 지점부터 이어서 실행해요.

이 방식이 자원 효율이 가장 좋아요. Triggerer 프로세스 하나는 수천 개의 조건을 동시에 지켜볼 수 있으니까, 나머지 워커 풀은 다른 태스크에 남겨둘 수 있죠.

트레이드오프는 다음과 같아요.

• Triggerer 컴포넌트가 실행 중이어야 해요. Triggerer가 없는 배포에서는 이 패턴을 쓸 수 없어요.
• 커스텀 지연 가능 연산자를 작성하려면 연산자 외에도 전용 Trigger 클래스를 구현해야 해요.
• 폴링 로직은 Triggerer의 async 이벤트 루프 안에서 돌아요. Trigger 안에서 블로킹 호출을 하면 Triggerer 프로세스 전체가 멈춰버려요.

이미 지연 가능 연산자가 존재하거나, 팀이 구현에 익숙하다면 자원 효율 때문에 이 경로가 권장돼요. 자세한 내용은 Deferrable Operators & Triggers를 참고하세요.

재개 가능한 태스크 (Resumable Tasks)

재개 가능한 태스크는 진행을 잃기 전에 체크포인트를 태스크 상태 저장소(task state store)에 저장해둬요. 재시도할 때 태스크는 그 체크포인트를 읽고, 처음부터가 아니라 멈췄던 지점부터 이어가요.

워커 슬롯은 일반 동기 연산자와 마찬가지로 태스크 전체 시간 동안 잡혀 있어요. 즉 이 방식의 이점은 자원 효율이 아니라 크래시 안전성과 연속성이에요.

재개 가능한 태스크는 이럴 때 유용해요.

• 외부 시스템용 지연 가능 연산자가 없고, 직접 작성하기도 현실적이지 않을 때
• Triggerer를 운영하지 않고 크래시 복구를 원할 때
• 태스크가 점진적으로 작업을 처리할 때(예: 파일 목록을 읽거나 API 결과를 페이지네이션) 마지막 완료 배치부터 재개할 수 있게 하려면

일반 패턴

태스크는 시작할 때 task_state_store에서 체크포인트를 읽고, 작업을 하고, 업데이트된 체크포인트를 쓴 뒤 계속하거나 끝내요. 다음 실행(크래시 후 재시도든 의도적 재스케줄이든)에서는 다시 체크포인트를 읽고 그 지점부터 이어가요.

from airflow.sdk import dag, task

@dag(schedule=None)
def process_files_dag():
    @task(retries=5)
    def process_files(context=None):
        task_state_store = context["task_state_store"]
        files = ["a.csv", "b.csv", "c.csv", "d.csv"]
        last_processed = task_state_store.get("last_processed")
        start_index = 0
        if last_processed is not None:
            start_index = files.index(last_processed) + 1
        for file in files[start_index:]:
            # ... process the file ...
            task_state_store.set("last_processed", file)

    process_files()

process_files_dag()

이 패턴은 추가 작업 없이 context만으로 동작해요. 상태 저장소는 태스크 인스턴스 범위의 단순한 키-값 저장소이고, 무엇을 체크포인트할지는 전적으로 여러분에게 달려 있어요.

외부 잡용 재개 가능 연산자

태스크가 외부 시스템에 잡을 제출하고 완료를 폴링하는 경우에는 추가 문제가 있어요. 재시도 시 첫 잡이 아직 실행 중일 수 있는데도 태스크가 두 번째 잡을 제출할 수 있다는 거죠. ResumableJobMixin은 폴링을 시작하기 전에 외부 잡 식별자를 영속화해서, 재시도 시 새 잡을 제출하는 대신 기존 잡에 다시 연결하도록 해줘요. 이걸 수동으로 처리할 필요가 없어지죠.

더 자세한 내용과 동작 예시는 ResumableJobMixin을 참고하세요.

비동기 태스크 (Asynchronous Tasks)

참고로 async 태스크 지원은 Python 태스크에만 적용돼요. @task 데코레이터를 쓴 async def 함수와 BaseAsyncOperator를 상속한 클래스 기반 연산자만 해당되고, 다른 연산자 타입에는 사용할 수 없어요.

Python 태스크는 async/await 문법을 지원해요. 데코레이트된 콜러블이 async 함수면 Airflow가 이를 이벤트 루프 안에서 실행해요. 그 결과 한 번의 태스크 실행 안에서 HTTP 요청, DB 쿼리, 파일 읽기 같은 많은 동시 I/O 작업을, 각각을 기다리며 이벤트 루프를 막지 않고 펼쳐서(fan-out) 실행할 수 있어요.

워커 슬롯은 태스크 전체 시간 동안 잡혀 있어요. async 태스크는 긴 외부 대기나 크래시 복구를 위해 설계된 게 아니라, 한 번의 실행 안에서 끝나는 고처리량 I/O 작업을 위해 설계됐어요.

언제 async 태스크를 쓰고 언제 지연 가능 연산자를 쓸지에 대한 안내는 Deferred vs Async Operators를 참고하세요.

세 방식 비교

더 알아보기

• Deferrable Operators & Triggers
• Task State Store - 체크포인트 저장소
• ResumableJobMixin API