Airflow

Task, Operator, Sensor, TaskFlow의 관계

Task는 Airflow에서 실행의 기본 단위다. 태스크는 Dag 안에 배치되고, 실행 순서를 표현하기 위해 서로 업스트림·다운스트림 의존성을 갖는다. 태스크에는 세 가지 기본 종류가 있다: Operator(빠르게 조합해 Dag의 대부분을 구성할 수 있는 사전 정의된 태스크 템플릿), Sensor(외부 이벤트가 일어나기를 기다리는 것에 특화된 Operator의 특수 서브클래스), 그리고 TaskFlow ``@task``로 데코레이트된, 커스텀 Python 함수를 패키징한 태스크다. 내부적으로 이들은 모두 Airflow의 ``BaseOperator``의 서브클래스이며 Task와 Operator 개념은 다소 상호 교환 가능하다 — 다만 개념적으로는 Operator/Sensor를 '템플릿'으로, Dag 파일에서 그것을 호출하는 순간 만들어지는 것을 'Task'로 구분해 생각하는 것이 유용하다.

Task의 핵심은 서로 간의 관계 — 업스트림/다운스트림 의존성 — 를 정의하는 것이다. 태스크를 먼저 선언한 뒤 의존성을 나중에 선언한다. 업스트림 태스크란 다른 태스크를 직접 선행하는 태스크를 말하며(과거에는 parent 태스크라고 불렀다), 계층상 더 상위에 있는(직접 부모가 아닌) 태스크를 가리키는 개념이 아니다. downstream도 마찬가지로 직접적인 자식(child)이어야 한다. 의존성은 ``>>``/``<<`` 비트시프트 연산자나 ``set_upstream``/``set_downstream`` 메서드로 선언하며, 둘은 동일하게 동작하지만 비트시프트 연산자가 대체로 더 읽기 쉬워 권장된다.

기본적으로 태스크는 모든 업스트림(부모) 태스크가 성공해야 실행되지만, 분기·일부 업스트림만 기다리기·히스토리상 위치에 따른 동작 변경 등 다양한 방법으로 이 동작을 바꿀 수 있다. 태스크는 기본적으로 서로 정보를 전달하지 않고 완전히 독립적으로 실행되며, 태스크 간 정보를 전달하려면 XCom을 써야 한다.

Task Instance와 상태 목록

Dag가 실행될 때마다 Dag Run으로 인스턴스화되듯, 그 Dag 아래의 태스크들도 함께 Task Instance로 인스턴스화된다. Task의 인스턴스는 주어진 Dag(따라서 주어진 데이터 인터벌)에 대한 그 태스크의 특정 실행을 나타내며, 태스크가 생명주기의 어느 단계에 있는지를 나타내는 '상태(state)'를 갖는 존재이기도 하다.

Task Instance가 가질 수 있는 상태는 다음과 같다: ``none``(아직 실행 대기열에 들어가지 않음, 의존성이 아직 충족되지 않음), ``scheduled``(스케줄러가 의존성이 충족되었다고 판단해 실행되어야 함), ``queued``(익스큐터에 할당되어 워커를 기다리는 중), ``running``(워커에서 실행 중, 또는 로컬/동기 익스큐터에서 실행 중), ``success``(오류 없이 실행 완료), ``restarting``(실행 중일 때 외부에서 재시작이 요청됨), ``failed``(실행 중 오류가 발생해 실행 실패), ``skipped``(분기·LatestOnly 등으로 스킵됨), ``upstream_failed``(업스트림 태스크가 실패했고 Trigger Rule이 그 업스트림을 필요로 함), ``up_for_retry``(실패했지만 남은 재시도 횟수가 있어 재스케줄됨), ``up_for_reschedule``(reschedule 모드인 Sensor가 대기 중), ``deferred``(트리거러에 지연됨), ``awaiting_input``(Human-in-the-loop 태스크가 사람의 응답을 기다리는 중이며, 스케줄러가 관리하고 워커 슬롯이나 트리거러를 사용하지 않음), ``removed``(Run이 시작된 이후 Dag에서 사라짐).

이상적으로 태스크는 ``none`` → ``scheduled`` → ``queued`` → ``running`` → ``success`` 순서로 흘러간다. 커스텀 Task(Operator)가 실행될 때는 태스크 인스턴스의 복사본을 전달받으며, 태스크 메타데이터를 조회하는 것 외에도 XCom 관련 메서드 등을 사용할 수 있다.

Timeout, SLA, 특수 예외

태스크에 최대 실행 시간을 두고 싶다면 ``execution_timeout`` 속성을 최대 허용 실행 시간을 나타내는 ``datetime.timedelta`` 값으로 설정하면 된다. 이는 센서를 포함한 모든 Airflow 태스크에 적용되며, 매 실행(execution)마다 허용되는 최대 시간을 제어한다. 이를 넘기면 ``AirflowTaskTimeout``이 발생한다.

추가로 센서에는 ``timeout`` 파라미터가 있는데, 이는 오직 ``reschedule`` 모드의 센서에만 의미가 있다. ``timeout``은 센서가 성공하기까지 허용되는 최대 시간을 제어하며, 이를 넘기면 ``AirflowSensorTimeout``이 발생해 센서가 재시도 없이 즉시 실패한다. 예를 들어 SFTP 서버를 60초 간격으로 폴링하는 reschedule 모드 센서에서, 매번의 poke는 ``execution_timeout=60초``까지 허용되고, poke 자체가 60초를 넘기면 ``AirflowTaskTimeout``이 발생하되 ``retries``만큼 재시도할 수 있다. 반면 첫 실행부터 파일이 나타날 때까지 전체적으로 허용되는 시간은 ``timeout=3600초``이며, 이 시간 안에 파일이 나타나지 않으면 ``AirflowSensorTimeout``이 발생하고 이때는 재시도하지 않는다. 네트워크 장애 등 다른 이유로 실패한다면 ``retries``만큼 재시도할 수 있지만, 재시도해도 ``timeout``이 리셋되지는 않으며 처음 시작부터 총 3600초 이내에 성공해야 한다는 제약은 그대로 유지된다.

Airflow 2의 SLA 기능은 3.0에서 제거되었고, Airflow 3.1에서 Deadline Alerts로 대체되었다.

커스텀 Task/Operator 코드 안에서 태스크 상태를 직접 제어하고 싶다면 두 가지 특수 예외를 raise할 수 있다: ``AirflowSkipException``은 현재 태스크를 skipped로 표시하고, ``AirflowFailException``은 남은 재시도 횟수를 무시하고 현재 태스크를 즉시 failed로 표시한다. 예를 들어 데이터가 없다는 것을 코드가 스스로 알고 있어 더 빨리 스킵하고 싶거나, API 키가 유효하지 않다는 것을 감지해 재시도로는 해결되지 않을 오류를 더 빨리 실패시키고 싶을 때 유용하다.