Airflow

Dag 정의 파일은 "설정 파일"이다

Airflow의 Python 스크립트는 코드처럼 보이지만, 실제로는 Dag의 구조를 기술하는 설정 파일에 가깝다고 생각하는 것이 정확하다. 여기서 정의한 태스크들은 이 스크립트가 실행되는 환경과는 다른 환경에서 실제로 실행된다 — 즉 이 스크립트 자체는 데이터를 처리하기 위한 곳이 아니다. 이 스크립트의 주된 역할은 Dag 객체를 정의하는 것이고, 이 파일은 Dag File Processor가 변경 사항이 있는지 주기적으로 검사하기 때문에 빠르게 평가(evaluate)되어야 한다. 만약 이 파일 최상위 레벨에 무거운 연산이나 외부 API 호출을 넣어두면, 매 파싱 주기마다 그 코드가 다시 실행되어 스케줄러 성능에 악영향을 줄 수 있다.

스크립트 작성의 첫 단계는 필요한 라이브러리를 임포트하는 것으로, 다른 Python 스크립트와 다르지 않다. Python과 Airflow가 모듈을 어떻게 다루는지에 대한 더 깊은 내용은 별도의 모듈 관리 문서에서 다룬다.

default_args와 태스크 인자의 우선순위

Dag과 태스크를 만들 때, 각 태스크에 인자를 직접 전달할 수도 있고 딕셔너리 형태의 기본 파라미터 집합(``default_args``)을 한 번에 정의해 재사용할 수도 있다. 보통은 후자가 더 효율적이고 코드도 깔끔해진다. ``BaseOperator``가 받는 파라미터의 전체 목록은 ``airflow.sdk.BaseOperator`` 문서에서 확인할 수 있다.

태스크 인자가 적용되는 우선순위는 다음과 같다: 1) 태스크에 명시적으로 전달된 인자, 2) ``default_args`` 딕셔너리에 정의된 값, 3) 오퍼레이터 자체의 기본값(있는 경우). 즉 명시적으로 지정한 값이 항상 default_args보다 우선하고, default_args가 없으면 오퍼레이터의 기본값이 적용된다.

모든 태스크는 ``task_id``와 ``owner``를 반드시 갖고 있어야 하며, 그렇지 않으면 Airflow가 에러를 발생시킨다. 다만 새로 설치한 Airflow는 ``owner``의 기본값을 ``"airflow"``로 설정해두므로, 실제로 사용자가 신경 써야 할 것은 거의 항상 ``task_id``뿐이다.

오퍼레이터로 태스크 정의하기

오퍼레이터(Operator)는 Airflow에서 하나의 작업 단위를 나타내며, 워크플로우를 구성하는 빌딩 블록이다. 오퍼레이터를 이용해 실제로 어떤 작업이 수행될지를 정의한다. 모든 오퍼레이터는 태스크를 실행하는 데 필요한 핵심 인자들을 담고 있는 ``BaseOperator``에서 파생된다. 대표적인 오퍼레이터로는 ``PythonOperator``, ``BashOperator``, ``KubernetesPodOperator`` 등이 있으며, 이 튜토리얼에서는 간단한 bash 명령을 실행하기 위해 ``BashOperator``를 사용한다. (Airflow는 더 Python다운 방식으로 워크플로우를 정의하는 TaskFlow API도 제공하는데, 이는 다음 튜토리얼에서 다룬다.)

오퍼레이터를 실제로 사용하려면 그것을 태스크로 인스턴스화해야 한다. 태스크는 그 오퍼레이터가 Dag의 컨텍스트 안에서 어떻게 작업을 수행할지를 결정한다. 예를 들어 두 개의 서로 다른 bash 스크립트를 실행하기 위해 ``BashOperator``를 두 번 인스턴스화할 수 있으며, 이때 각각을 구분해주는 고유 식별자가 ``task_id``다. 오퍼레이터 고유 인자(예: ``bash_command``)와 ``BaseOperator``에서 상속된 공통 인자(예: ``retries``)를 함께 섞어 쓸 수 있다는 점도 특징이다 — 예를 들어 두 번째 태스크에서만 ``retries``를 3으로 오버라이드하는 식이다.

Jinja 템플릿과 태스크/Dag 문서화

Airflow는 Jinja 템플릿 엔진의 힘을 빌려, 내장 파라미터와 매크로에 접근할 수 있게 해준다. 가장 흔히 쓰이는 템플릿 변수는 오늘 날짜를 나타내는 ``{{ ds }}``다. 템플릿 안에서는 ``{% %}`` 블록으로 로직을 넣을 수도 있고, ``bash_command='templated_command.sh'``처럼 파일을 직접 참조할 수도 있어 코드를 더 깔끔하게 정리할 수 있다. ``user_defined_macros``와 ``user_defined_filters``를 정의하면 자신만의 변수와 필터를 템플릿에서 쓸 수도 있다. 템플릿에서 참조할 수 있는 변수와 매크로 전체 목록은 별도의 templates-ref 문서에서 확인할 수 있다.

Dag이나 개별 태스크에는 문서를 추가할 수 있다. Dag 문서는 UI에서 Markdown으로 렌더링되며, 태스크 문서는 일반 텍스트, Markdown, reStructuredText, JSON, YAML 중 하나로 작성할 수 있다. ``doc_md``를 태스크 문서화에 사용하면 Airflow는 인라인 코드, 코드 블록, KaTeX로 렌더링되는 ``math`` 펜스, Mermaid 다이어그램을 렌더링하는 ``mermaid`` 펜스 같은 일반적인 Markdown 기능을 지원한다. 이런 문서는 Task Instance Details 페이지에 표시되며, 문서를 그 태스크를 설명하는 코드 바로 옆에 두는 것이 좋은 습관이다.

태스크 의존성 정의하기

Airflow에서 태스크는 서로 의존할 수 있다. 예를 들어 ``t1``, ``t2``, ``t3``라는 세 태스크가 있다면 여러 방식으로 의존성을 정의할 수 있다.

t1.set_downstream(t2)
# t2가 t1의 성공적인 실행에 의존한다는 뜻이며, 아래와 동일하다:
t2.set_upstream(t1)

# 비트 시프트 연산자로도 체이닝할 수 있다:
t1 >> t2
# 업스트림 방향도 마찬가지다:
t2 << t1

# 여러 의존성을 연쇄적으로 표현할 수 있다:
t1 >> t2 >> t3

# 태스크 리스트도 의존성으로 지정할 수 있다 — 아래는 모두 같은 효과를 낸다:
t1.set_downstream([t2, t3])
t1 >> [t2, t3]
[t2, t3] << t1

Airflow는 Dag 안에서 순환(cycle)이 감지되거나 같은 의존성이 중복해서 참조되면 에러를 발생시킨다는 점을 유의해야 한다 — Dag은 이름 그대로 비순환 그래프여야 하기 때문이다.