태스크 (Tasks)
태스크 (Tasks)
Airflow에서 태스크(Task)는 실행의 기본 단위예요. 태스크들은 DAG 안에 배치되고, 실행 순서를 표현하기 위해 서로 업스트림(upstream), 다운스트림(downstream) 의존 관계를 맺게 됩니다.
태스크에는 크게 세 가지 종류가 있어요.
- 오퍼레이터(Operators): 미리 정의된 태스크 템플릿으로, DAG의 대부분을 빠르게 이어 붙여서 만들 수 있어요.
- 센서(Sensors): 오퍼레이터의 특수한 하위 클래스로, 외부 이벤트가 발생할 때까지 기다리는 것만 담당해요.
- TaskFlow 데코레이터가 붙은
@task: 커스텀 파이썬 함수를 태스크로 포장한 거예요.
사실 이들은 내부적으로 모두 Airflow의 BaseOperator 하위 클래스라서, 태스크와 오퍼레이터의 개념은 어느 정도 서로 교환 가능합니다. 그래도 둘을 별개의 개념으로 생각하는 게 좋아요. 오퍼레이터와 센서는 본질적으로 템플릿이고, DAG 파일에서 그것을 호출하는 순간 태스크가 만들어집니다.
관계 (Relationships)
태스크를 쓰는 핵심은 태스크들이 서로 어떻게 연결되는지, 즉 의존 관계(dependencies) 를 정의하는 일이에요. Airflow에서는 이를 업스트림과 다운스트림 태스크라고 부릅니다. 태스크를 먼저 선언하고, 그 다음에 의존 관계를 선언하는 순서로 진행해요.
참고
우리는 다른 태스크 바로 앞에 오는 태스크를 업스트림이라고 불러요. 예전에는 부모(parent) 태스크라고 부르기도 했죠. 다만 이 개념은 태스크 계층에서 더 높은 위치에 있는 태스크를 가리키는 게 아니라는 점을 주의해야 해요 (즉, 직접적인 부모가 아니라는 뜻이에요). 다운스트림 태스크도 같은 정의가 적용돼요. 다른 태스크의 직접적인 자식이어야 합니다.
의존 관계를 선언하는 방법은 두 가지예요. >>와 << (비트 시프트) 연산자를 쓰는 방법:
first_task >> second_task >> [third_task, fourth_task]
또는 더 명시적인 set_upstream과 set_downstream 메서드를 쓰는 방법:
first_task.set_downstream(second_task)
third_task.set_upstream(second_task)
두 방법은 정확히 같은 일을 하지만, 일반적으로는 비트 시프트 연산자를 권장해요. 대부분의 경우 읽기가 더 쉽기 때문이죠.
기본적으로 태스크는 모든 업스트림(부모) 태스크가 성공했을 때 실행됩니다. 하지만 이런 동작을 바꾸는 방법은 아주 많아요. 분기(branching)를 추가하거나, 일부 업스트림 태스크만 기다리게 하거나, 현재 실행이 히스토리에서 어느 시점인지에 따라 동작을 바꿀 수도 있어요. 더 자세한 내용은 제어 흐름(Control Flow)을 참고하세요.
태스크들은 기본적으로 서로에게 정보를 전달하지 않으며, 완전히 독립적으로 실행됩니다. 한 태스크에서 다른 태스크로 정보를 전달하고 싶다면 XComs를 사용해야 해요.
태스크 인스턴스 (Task Instances)
DAG가 실행될 때마다 DAG Run으로 인스턴스화되는 것과 마찬가지로, DAG 아래의 태스크들도 *태스크 인스턴스(Task Instance)*로 인스턴스화됩니다.
태스크 인스턴스는 특정 DAG(따라서 특정 데이터 구간)에 대해 해당 태스크의 특정 실행을 의미해요. 또한 태스크가 현재 수명 주기의 어느 단계에 있는지를 나타내는 상태(state) 를 가지는 태스크의 표현이기도 합니다.
태스크 인스턴스가 가질 수 있는 상태는 다음과 같아요.
none: 태스크가 아직 실행을 위해 큐에 들어가지 않은 상태 (의존 관계가 아직 충족되지 않음)scheduled: 스케줄러가 태스크의 의존 관계가 충족되었고 실행되어야 한다고 판단한 상태queued: 태스크가 Executor에 배정되었고 워커를 기다리고 있는 상태running: 태스크가 워커(또는 로컬/동기 Executor)에서 실행 중인 상태success: 태스크가 오류 없이 끝까지 실행된 상태restarting: 실행 중이던 태스크가 외부에서 재시작을 요청받은 상태failed: 실행 중 오류가 발생해 태스크가 실패한 상태skipped: 분기(branching), LatestOnly 등으로 인해 태스크가 건너뛰어진 상태upstream_failed: 업스트림 태스크가 실패했고 트리거 규칙(Trigger Rule)이 그걸 필요로 하는 상태up_for_retry: 태스크가 실패했지만 재시도 횟수가 남아 있어 다시 스케줄될 상태up_for_reschedule:reschedule모드인 센서 상태deferred: 태스크가 트리거로 연기(deferred)된 상태awaiting_input: Human-in-the-loop 태스크로 사람의 응답을 기다리는 상태. 스케줄러가 관리하며 워커 슬롯이나 트리거러를 사용하지 않아요.removed: 실행이 시작된 후 태스크가 DAG에서 사라진 상태
이상적으로 태스크는 none → scheduled → queued → running 순서로 흘러 마지막에 success에 도달해야 해요.
커스텀 태스크(오퍼레이터)가 실행되는 동안에는 태스크 인스턴스의 사본을 전달받게 됩니다. 태스크 메타데이터를 조사할 수 있을 뿐만 아니라, XComs 같은 기능을 위한 메서드도 포함하고 있어요.
관계 용어 (Relationship Terminology)
어떤 태스크 인스턴스든 다른 인스턴스와 맺는 관계에는 두 가지 유형이 있어요.
첫째, 업스트림과 다운스트림 태스크 관계입니다:
task1 >> task2 >> task3
DAG가 실행되면 서로 업스트림/다운스트림인 이 태스크들 각각에 인스턴스가 생성되는데, 모두 같은 데이터 구간을 가집니다.
또한 같은 태스크의 인스턴스가 존재할 수 있는데, 다른 데이터 구간에 대한 것이에요. 같은 DAG의 다른 실행에서 나온 것들이죠. 이런 관계를 우리는 이전(previous) 과 다음(next) 이라고 부릅니다. 업스트림/다운스트림과는 완전히 다른 관계예요!
참고
오래된 Airflow 문서 중에는 여전히 "previous"를 "upstream"이라는 뜻으로 쓰는 경우가 있어요. 이런 곳을 발견하면 고쳐주시면 감사하겠습니다!
타임아웃 (Timeouts)
태스크에 최대 실행 시간을 설정하고 싶다면 execution_timeout 속성에 최대 허용 실행 시간을 나타내는 datetime.timedelta 값을 지정하면 돼요. 이는 센서를 포함한 모든 Airflow 태스크에 적용됩니다. execution_timeout은 매 실행마다 허용되는 최대 시간을 제어해요. execution_timeout이 초과되면 태스크가 타임아웃되고 AirflowTaskTimeout이 발생합니다.
추가로 센서에는 timeout 파라미터가 있어요. 이 값은 reschedule 모드의 센서에만 영향을 줍니다. timeout은 센서가 성공하기까지 허용되는 최대 시간을 제어해요. timeout이 초과되면 AirflowSensorTimeout이 발생하고, 센서는 재시도 없이 즉시 실패합니다.
다음 SFTPSensor 예시가 이를 잘 보여줘요. reschedule 모드의 센서는 성공할 때까지 주기적으로 실행되고 재스케줄됩니다.
- 센서가 SFTP 서버를 poke할 때마다
execution_timeout으로 정의된 대로 최대 60초가 허용돼요. - 센서가 SFTP 서버를 poke하는 데 60초 이상 걸리면
AirflowTaskTimeout이 발생해요. 이 경우 센서는 재시도할 수 있는데,retries로 정의된 대로 최대 2번 재시도할 수 있어요. - 첫 실행 시작부터 (즉 'root/test' 파일이 나타난 뒤) 최종적으로 성공할 때까지, 센서는
timeout으로 정의된 대로 최대 3600초가 허용돼요. 즉 3600초 안에 파일이 SFTP 서버에 나타나지 않으면 센서가AirflowSensorTimeout을 발생시키는 거죠. 이 오류가 발생하면 재시도하지 않습니다. - 3600초 구간 동안 네트워크 장애 같은 다른 이유로 센서가 실패하면
retries로 정의된 대로 최대 2번 재시도할 수 있어요. 재시도는timeout을 초기화하지 않습니다. 성공하기까지 여전히 총 3600초가 주어져요.
sensor = SFTPSensor(
task_id="sensor",
path="/root/test",
execution_timeout=timedelta(seconds=60),
timeout=3600,
retries=2,
mode="reschedule",
)
SLA
Airflow 2의 SLA 기능은 3.0에서 제거되었고, Airflow 3.1에서는 Deadlines Alerts로 대체되었어요.
특수 예외 (Special Exceptions)
커스텀 태스크/오퍼레이터 코드 안에서 태스크의 상태를 직접 제어하고 싶다면, Airflow가 제공하는 두 가지 특수 예외를 발생시키면 됩니다.
AirflowSkipException은 현재 태스크를 skipped로 표시해요.AirflowFailException은 남은 재시도 횟수를 무시하고 현재 태스크를 failed로 표시해요.
이 예외들은 코드가 자기 환경에 대한 추가 지식을 알고 있고 더 빨리 실패/건너뛰기를 원할 때 유용해요. 예를 들어 데이터가 없다는 걸 미리 알면 건너뛰거나, API 키가 유효하지 않다는 걸 감지하면 빨리 실패하는 식이죠 (키가 잘못된 건 재시도로 해결되지 않으니까요).
재시도 정책 (Retry Policies)
기본적으로 Airflow는 오류 유형과 관계없이 고정된 횟수와 지연으로 실패한 태스크를 재시도해요. 재시도 정책(retry policy) 을 쓰면 태스크 코드를 수정하지 않고도 어떤 태스크나 오퍼레이터의 파라미터로 예외별 재시도 동작을 설정할 수 있습니다.
예외 유형을 동작에 매핑하는 규칙으로 정책을 정의하고, 이를 태스크에 적용하면 돼요:
from airflow.sdk import DAG, ExceptionRetryPolicy, RetryAction, RetryRule, task
API_RETRY_POLICY = ExceptionRetryPolicy(
rules=[
RetryRule(
exception="requests.exceptions.HTTPError",
action=RetryAction.RETRY,
retry_delay=timedelta(minutes=5),
reason="Rate limit, backing off",
),
RetryRule(
exception="google.auth.exceptions.RefreshError",
action=RetryAction.FAIL,
reason="Auth failure, not retryable",
),
RetryRule(
exception=ConnectionError,
action=RetryAction.RETRY,
retry_delay=timedelta(seconds=30),
),
],
)
동작 방식 (How it works)
정책은 태스크 워커 프로세스에서 실행돼요 (스케줄러에서는 절대 실행되지 않습니다). 예외를 잡은 뒤 태스크의 다음 상태를 결정하기 전에 실행되죠. 정책의 각 결정은 태스크 로그에 Retrypolicydecisionaction=<action>reason=<reason> 형식으로 기록됩니다.
태스크가 실패하면 정책이 예외를 평가하고 세 가지 동작 중 하나를 반환해요.
- RETRY – 태스크를 재시도하며, 선택적으로
retry_delay를 덮어쓰는 커스텀 지연을 사용할 수 있어요. 재시도는 여전히 태스크의retries횟수에 제약을 받아요. 정책이 더 일찍 실패할 수는 있지만, 설정된 최대치를 넘어 확장할 수는 없습니다. - FAIL – 남은 재시도를 건너뛰고 즉시 실패해요.
- DEFAULT – 표준 재시도 로직(
retries횟수와retry_delay)으로 넘어가요.
규칙은 순서대로 평가되며, 첫 번째로 매칭된 규칙이 우선해요. 매칭되는 규칙이 없으면 정책은 DEFAULT(표준 재시도 동작)를 반환합니다.
예외 매칭 (Exception matching)
예외 유형은 파이썬 클래스나 점(dot)으로 구분된 import 경로 문자열(예: "requests.exceptions.HTTPError")로 지정할 수 있어요. 문자열 경로는 DAG 파싱 시점에 검증되는데, 점이 없는 경로는 즉시 ValueError를 발생시키고, 해석할 수 없는 경로는 경고를 만들어요.
기본적으로 규칙은 isinstance 매칭을 사용하므로, OSError에 대한 규칙은 하위 클래스인 ConnectionError에도 매칭됩니다. 정확한 타입 매칭을 원하면 match_subclasses=False로 설정하세요:
RetryRule(exception=OSError, match_subclasses=False) # OSError만 매칭, 하위 클래스는 제외
기존 파라미터와의 조합 (Composition with existing parameters)
| 파라미터 | retry_policy가 설정되었을 때의 동작 |
|---|---|
retries |
여전히 최대 재시도 횟수. 정책이 더 일찍 실패할 수는 있지만 이를 초과할 수는 없어요. |
retry_delay / retry_exponential_backoff / max_retry_delay |
정책이 DEFAULT를 반환하거나 RetryDecision.retry_delay가 None일 때 사용돼요. |
on_retry_callback |
정책에 의한 재시도를 포함한 모든 재시도에서 실행돼요. |
AirflowFailException |
항상 우선순위를 가져요. 이 예외에 대해서는 정책이 결코 조회되지 않습니다. |
재사용 가능한 정책 (Reusable policies)
정책을 한 번 정의해 두면 default_args나 공용 모듈을 통해 여러 DAG에서 공유할 수 있어요:
# policies.py -- 어떤 DAG에서든 import
STANDARD_RETRY_POLICY = ExceptionRetryPolicy(
rules=[
RetryRule(exception="requests.exceptions.HTTPError", action=RetryAction.FAIL),
RetryRule(exception=ConnectionError, retry_delay=timedelta(seconds=10)),
],
)
매핑 태스크 (Mapped tasks)
정책은 .partial()을 통한 동적 태스크 매핑에서도 동작해요. 정책은 매핑된 태스크 인스턴스 각각에 적용됩니다. 10개 중 2번째 인스턴스가 FAIL에 걸리면 나머지 9개는 독립적으로 계속 실행돼요:
@task.partial(retry_policy=my_policy).expand(input=[1, 2, 3])
def my_mapped_task(input): ...
정책은 .partial()을 통해 태스크 레벨에서 설정되며, 모든 매핑된 인스턴스가 하나의 정책을 공유해요. .expand()에서는 인덱스별 변형이 지원되지 않지만, 정책의 evaluate() 메서드가 예외, try_number, 전체 컨텍스트를 받기 때문에 필요하면 정책 안에서 인덱스별 분기를 할 수 있어요.
커스텀 재시도 정책 (Custom retry policies)
고급 사례에서는 RetryPolicy를 서브클래싱하고 evaluate()를 구현하면 돼요. 선언적인 ExceptionRetryPolicy 규칙이 포착할 수 없는 예외 속성(상태 코드, 응답 헤더, 페이로드)을 조사해야 할 때 서브클래싱이 올바른 선택이에요. 이 메서드는 예외, 시도 횟수, 최대 시도 횟수, 그리고 전체 Airflow 컨텍스트(dag_run, params 등)를 받습니다:
from datetime import timedelta
import requests
from airflow.sdk import RetryDecision, RetryPolicy
class HTTPStatusRetryPolicy(RetryPolicy):
"""HTTP 상태 코드로 재시도 결정을 라우팅하고, 429에서 Retry-After를 존중."""
def evaluate(self, exception, try_number, max_tries, context=None):
if isinstance(exception, requests.HTTPError) and exception.response is not None:
status = exception.response.status_code
if status == 429: # rate limited -- Retry-After 헤더 존중
retry_after = int(exception.response.headers.get("Retry-After", 60))
return RetryDecision.retry(retry_delay=timedelta(seconds=retry_after))
if 500 <= status < 600: # server error -- 재시도 가치 있음
return RetryDecision.retry()
if 400 <= status < 500: # client error -- 재시도 불가
return RetryDecision.fail(reason=f"HTTP {status}")
return RetryDecision.default()
두 번째 패턴은 실행 context를 사용해 DAG가 어떻게 트리거되었는지에 따라 결정을 내려요:
from airflow.sdk import RetryDecision, RetryPolicy
class BackfillAwareRetryPolicy(RetryPolicy):
"""백필 중에는 빠르게 실패시켜 과거 오류가 즉시 드러나게 함."""
def evaluate(self, exception, try_number, max_tries, context=None):
if context and context["dag_run"].run_type == "backfill":
return RetryDecision.fail(reason="Backfill run -- not retrying")
return RetryDecision.default()
태스크 인스턴스 하트비트 타임아웃 (Task Instance Heartbeat Timeout)
완벽하게 돌아가는 시스템은 없고, 태스크 인스턴스도 가끔 죽게 마련이에요.
TaskInstance는 연관된 작업(job)이 비활성 상태인데도 running 상태에 갇힐 수 있어요 (예를 들어 TaskInstance 워커의 메모리가 부족해진 경우). 이런 태스크를 예전에는 좀비(zombie) 태스크라고 불렀어요. Airflow는 이런 태스크를 주기적으로 찾아내 정리하고, TaskInstance를 failed로 표시하거나 재시도 횟수가 남아 있으면 재시도합니다. TaskInstance의 하트비트는 여러 이유로 타임아웃될 수 있어요. 그중 일부는:
- Airflow 워커가 메모리가 부족해 OOMKilled된 경우
- Airflow 워커가 liveness 프로브에 실패해서 시스템(예: Kubernetes)이 워커를 재시작한 경우
- 시스템(예: Kubernetes)이 스케일 다운하면서 Airflow 워커를 한 노드에서 다른 노드로 옮긴 경우
로컬에서 태스크 인스턴스 하트비트 타임아웃 재현하기
개발/테스트 과정에서 태스크 인스턴스 하트비트 타임아웃을 재현하고 싶다면 아래 단계를 따라보세요:
- 로컬 Airflow 설정에 아래 환경 변수를 설정하세요 (또는 airflow.cfg에서 해당 설정 값을 조정해도 됩니다)
export AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_SEC=600
export AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_TIMEOUT=2
export AIRFLOW__SCHEDULER__TASK_INSTANCE_HEARTBEAT_TIMEOUT_DETECTION_INTERVAL=5
- 완료에 약 10분이 걸리는(즉 오래 실행되는) 태스크를 가진 DAG를 준비하세요. 예를 들어 아래와 같은 DAG를 사용할 수 있어요:
from airflow.sdk import dag
from airflow.providers.standard.operators.bash import BashOperator
from datetime import datetime
@dag(start_date=datetime(2021, 1, 1), schedule="@once", catchup=False)
def sleep_dag():
t1 = BashOperator(
task_id="sleep_10_minutes",
bash_command="sleep 600",
)
sleep_dag()
위 DAG를 실행하고 잠시 기다리세요. <task_instance_heartbeat_timeout> 초가 지나면 TaskInstance가 failed로 표시됩니다.
Executor 구성 (Executor Configuration)
일부 Executor는 선택적인 태스크별 구성을 허용해요. 예를 들어 KubernetesExecutor는 태스크를 실행할 이미지를 설정할 수 있게 해주죠.
이 작업은 태스크나 오퍼레이터의 executor_config 인자를 통해 이루어집니다. KubernetesExecutor에서 실행될 태스크의 Docker 이미지를 설정하는 예시를 보여줄게요:
MyOperator(...,
executor_config={
"KubernetesExecutor":
{"image": "myCustomDockerImage"}
}
)
executor_config에 전달할 수 있는 설정은 executor마다 다르므로, 무엇을 설정할 수 있는지 확인하려면 개별 executor 문서를 읽어보세요.