Task

상위: Airflow

요약

Task는 DAG를 구성하는 개별 작업 단위입니다. Operators를 통해 정의되며, 독립적으로 실행됩니다. Task는 고유한 task_id를 가지며, 재시도, 타임아웃, 우선순위 등 다양한 속성을 설정할 수 있습니다. Task 간 의존성을 통해 실행 순서를 제어합니다.

Task란?

• 워크플로우를 구성하는 개별 작업 단위
• DAG 내에서 하나의 노드로 존재
• Operator를 사용해 정의
• 독립적으로 실행되며 재시도 가능

Task의 역할

• ETL 작업 수행
• 데이터 변환 및 처리
• 머신러닝 모델 실행
• 파일 이동 및 관리
• API 호출
• SQL 쿼리 실행

Task LifeCycle

주요 상태

Task 주요 속성

필수 속성

선택 속성

Task 정의 예제

기본 Task 정의

from airflow.operators.bash import BashOperator

task_1 = BashOperator(
    task_id='print_date',
    bash_command='date',
    retries=3,
    retry_delay=timedelta(minutes=5),
)

Task 속성 상세 설정

from airflow.operators.python import PythonOperator
from datetime import timedelta

def my_function():
    print("Task executed!")

task_2 = PythonOperator(
    task_id='python_task',
    python_callable=my_function,
    retries=2,
    retry_delay=timedelta(minutes=3),
    execution_timeout=timedelta(minutes=10),
    priority_weight=5,
    depends_on_past=False,
)

Task Instance

Task Instance란?

• Task의 특정 실행 인스턴스
• (dag_id, task_id, execution_date) 조합으로 고유 식별
• 각 DAG Run마다 새로운 Task Instance 생성

Task Instance 속성

• state: 현재 실행 상태
• start_date: 실행 시작 시간
• end_date: 실행 종료 시간
• duration: 실행 소요 시간
• try_number: 현재 시도 횟수

Task 실행 방식

순차 실행 (Sequential)

task_a >> task_b >> task_c

• task_a 완료 후 task_b 실행
• task_b 완료 후 task_c 실행

병렬 실행 (Parallel)

task_start >> [task_a, task_b, task_c]

• task_start 완료 후 task_a, task_b, task_c 동시 실행
• 독립적인 작업들을 병렬로 처리

자세한 내용은 Task Dependencies 참조

Task 간 데이터 공유

XCom 사용

• Task 간 소량의 데이터 전달
• XCom 참조

외부 스토리지 사용

• 대용량 데이터는 S3, GCS 등에 저장
• Task 간 파일 경로만 전달

주의사항

Task는 독립적이어야 함

• 각 Task는 독립적으로 실행 가능해야 함
• 다른 Task의 메모리나 상태에 의존 금지

Task는 멱등성을 가져야 함

• 여러 번 실행해도 같은 결과
• 재시도 시 부작용 없어야 함

Task는 원자적이어야 함

• 하나의 명확한 작업만 수행
• 너무 많은 작업을 하나의 Task에 넣지 말 것

다음 단계

• Operators - Task를 실행하는 다양한 Operator
• Task Dependencies - Task 간 의존성 설정
• Trigger Rules - Task 실행 조건 제어
• XCom - Task 간 데이터 전달

참고 자료

• Task 공식 문서
• Task Instance API