Airflow

충돌하는 의존성을 격리하는 네 가지 방법

때로는 일부 태스크가 Dag의 나머지 부분과는 다른 Python 의존성 — 특수한 라이브러리나 시스템 레벨 패키지 — 을 필요로 한다. TaskFlow는 이런 의존성을 격리하기 위해 여러 실행 환경을 지원한다.

**동적으로 생성되는 virtualenv**는 태스크 실행 시점에 임시 가상환경을 만든다. 실험적이거나 동적인 태스크에 적합하지만, 콜드 스타트 오버헤드가 발생할 수 있다.

**외부 Python 환경**은 미리 설치되어 있는 Python 인터프리터를 이용해 태스크를 실행한다. 일관된 환경이나 공유 가상환경이 필요할 때 이상적이다.

**Docker 환경**은 태스크를 Docker 컨테이너 안에서 실행한다. 태스크에 필요한 모든 것을 패키징하는 데 유용하지만, 워커에 Docker가 설치되어 있어야 한다. Airflow 2.2 이상과 Docker 프로바이더가 필요하다.

**KubernetesPodOperator**는 태스크를 Kubernetes 파드 안에서, 메인 Airflow 환경과 완전히 분리된 상태로 실행한다. 대형 태스크나 커스텀 런타임이 필요한 태스크에 이상적이다. Airflow 2.4 이상과 Kubernetes 프로바이더가 필요하다.

@task.sensor로 만드는 경량 센서

``@task.sensor``를 이용하면 Python 함수만으로 가볍고 재사용 가능한 센서를 만들 수 있다. 이렇게 만든 센서는 poke 모드와 reschedule 모드를 모두 지원한다 — poke 모드는 워커 슬롯을 계속 점유한 채 주기적으로 조건을 확인하고, reschedule 모드는 확인 사이 간격마다 워커 슬롯을 반납했다가 다시 스케줄링되는 방식으로 자원을 절약한다.

전통적인 태스크와 섞어 쓰기

데코레이팅된 태스크를 전통적인 오퍼레이터와 함께 조합해서 쓸 수도 있다. 이는 커뮤니티 프로바이더를 사용하거나, TaskFlow로 점진적으로 마이그레이션하는 과정에서 특히 유용하다. TaskFlow 태스크와 전통적인 태스크는 ``>>``로 체이닝하거나, ``.output`` 속성을 이용해 데이터를 주고받을 수 있다.

템플릿과 컨텍스트 변수 접근

전통적인 태스크와 마찬가지로, 데코레이팅된 TaskFlow 함수도 템플릿된 인자를 지원한다 — 파일에서 내용을 불러오거나 런타임 파라미터를 사용하는 것도 가능하다.

콜러블이 실행될 때 Airflow는 Jinja 템플릿에서 쓸 수 있는 것과 정확히 동일한 컨텍스트 키워드 인자 집합을 전달해준다. 함수에서 받고 싶은 컨텍스트 키를 키워드 인자로 명시하면 된다. 예를 들어 아래 콜러블은 ``ti``와 ``next_ds`` 컨텍스트 변수의 값을 받는다.

@task
def my_python_callable(*, ti, next_ds):
    pass

``**kwargs``로 전체 컨텍스트를 통째로 받을 수도 있지만, 이 경우 실제로 필요하지 않은 많은 항목까지 포함해 전체 컨텍스트를 펼쳐야 하므로 약간의 성능 저하가 발생할 수 있다. 따라서 앞서 보여준 것처럼 명시적인 인자를 쓰는 편이 권장된다.

@task
def my_python_callable(**kwargs):
    ti = kwargs["ti"]
    next_ds = kwargs["next_ds"]

때로는 태스크 콜러블에서 컨텍스트 변수를 직접 전달하지 않고, 호출 스택 깊숙한 곳에서 실행 컨텍스트에 접근하고 싶을 수도 있다. 이런 경우 ``get_current_context`` 메서드로 여전히 컨텍스트에 접근할 수 있다.

from airflow.sdk import get_current_context


def some_function_in_your_library():
    context = get_current_context()
    ti = context["ti"]

데코레이팅된 함수에 전달되는 인자는 자동으로 템플릿된다. ``templates_exts``를 이용하면 파일도 템플릿할 수 있다.

@task(templates_exts=[".sql"])
def read_sql(sql): ...

Dag 구조를 바꾸지 않는 조건부 실행

``@task.run_if()``나 ``@task.skip_if()``를 쓰면 Dag 구조 자체를 바꾸지 않고도, 런타임에 결정되는 동적인 조건에 따라 태스크의 실행 여부를 제어할 수 있다.

@task.run_if(lambda ctx: ctx["task_instance"].task_id == "run")
@task.bash()
def echo():
    return "echo 'run'"

이 패턴은 조건에 따라 아예 다른 브랜치로 분기하는 대신, 같은 위치의 태스크를 조건부로 건너뛰거나 실행하고 싶을 때 유용하다 — Dag의 시각적 구조(그래프)는 그대로 유지하면서 실행 여부만 런타임에 결정된다는 점이 브랜칭 오퍼레이터와의 차이다.