Airflow

TaskFlow API 개요

첫 번째 튜토리얼에서는 ``BashOperator`` 같은 전통적인 오퍼레이터로 첫 Dag을 만들었다. 이번에는 Airflow 2.0에서 도입된 **TaskFlow API**를 이용해 워크플로우를 작성하는, 더 현대적이고 Python다운 방식을 살펴본다.

TaskFlow API는 코드를 더 단순하고 깔끔하며 유지보수하기 쉽게 만드는 것을 목표로 한다. 일반 Python 함수를 작성하고 데코레이터를 붙이기만 하면, Airflow가 태스크 생성, 의존성 연결, 태스크 간 데이터 전달까지 나머지를 알아서 처리해준다. 이 튜토리얼에서는 TaskFlow API로 Extract → Transform → Load라는 간단한 ETL 파이프라인을 만들어본다.

@dag으로 Dag 정의하기

이전과 마찬가지로 Dag은 Airflow가 로드하고 파싱하는 Python 스크립트지만, 이번에는 ``@dag`` 데코레이터를 이용해 정의한다. Dag을 Airflow가 발견할 수 있게 하려면 ``@dag``으로 데코레이팅된 Python 함수를 호출해주기만 하면 된다.

2.4 버전부터는 ``@dag`` 데코레이터를 쓰거나 ``with`` 블록 안에서 Dag을 정의하는 경우, 더 이상 그 결과를 전역 변수에 할당할 필요가 없다 — Airflow가 자동으로 찾아준다. Dag이 로드된 후에는 Airflow UI의 Graph 뷰로 이동해 태스크가 어떻게 연결되어 있는지 시각적으로 확인할 수 있다.

@task로 태스크를 작성하고 자동으로 XCom 넘기기

TaskFlow에서는 각 태스크가 그냥 평범한 Python 함수다. ``@task`` 데코레이터를 붙이면 Airflow가 스케줄링하고 실행할 수 있는 태스크로 바뀐다. 함수의 반환값은 별도로 XCom을 다루지 않아도 자동으로 다운스트림 태스크에 전달된다 — 내부적으로 TaskFlow는 XCom을 이용해 이 데이터 전달을 자동으로 처리하며, 이전 방식에서 수동으로 XCom을 다루던 복잡함을 추상화해 감춘다.

``@task(multiple_outputs=True)``라고 지정하면, 그 함수가 여러 값을 담은 딕셔너리를 반환한다는 것을 Airflow에 알려주는 것이다. 이렇게 하면 반환된 딕셔너리의 각 키가 개별 XCom 항목이 되어, 다운스트림 태스크에서 특정 값만 골라 참조하기 쉬워진다. 만약 ``multiple_outputs=True``를 생략하면 딕셔너리 전체가 하나의 XCom으로 저장되고, 그 값을 쓸 때는 딕셔너리 전체를 통째로 꺼내야 한다.

옛날 방식과 비교: 수동 XCom 관리

TaskFlow API 이전에는 ``PythonOperator`` 같은 오퍼레이터를 쓰고, 태스크 사이에 ``XCom``을 이용해 데이터를 수동으로 전달해야 했다. 같은 Dag을 전통적인 방식으로 작성하면 각 함수 안에서 ``ti.xcom_pull(task_ids="extract")``처럼 명시적으로 XCom을 꺼내와야 하고, ``extract_task >> transform_task >> load_task``처럼 태스크 인스턴스 사이의 의존성도 별도로 연결해줘야 한다.

TaskFlow 방식은 이 모든 것을 자동으로 처리하지만, 겉으로 드러나지 않을 뿐 내부적으로는 여전히 XCom을 사용하고 의존성 그래프를 만든다 — 다만 그 복잡함이 추상화되어, 사용자는 비즈니스 로직에만 집중할 수 있다는 차이가 있다. TaskFlow의 반환값들은 자동으로 XCom으로 저장되며, 이 값들은 UI의 "XCom" 탭에서 확인할 수 있다. 전통적인 오퍼레이터에서 쓰던 수동 ``xcom_pull()``도 여전히 사용 가능하다.

에러 처리와 태스크 파라미터화

태스크 데코레이터에 재시도 횟수를 직접 지정해 재시도를 손쉽게 설정할 수 있다.

@task(retries=3)
def my_task(): ...

이렇게 하면 일시적인 장애가 태스크 실패로 곧바로 이어지지 않도록 도와준다.

또한 데코레이팅된 태스크는 여러 Dag에서 재사용할 수 있고, ``task_id``나 ``retries`` 같은 파라미터를 오버라이드할 수도 있다.

start = add_task.override(task_id="start")(1, 2)

심지어 데코레이팅된 태스크를 공유 모듈에서 임포트해 여러 Dag에서 재사용하는 것도 가능하다.