Airflow

Connection/Variable/XCom - Task Context와 airflow.sdk를 통한 접근

Hook과 Operator를 작성·확장할 때 Dag 작성자와 개발자는 다음 클래스들을 사용할 수 있다: 외부 서비스 자격증명·설정 접근을 제공하는 :class:`~airflow.sdk.Connection`, Airflow 설정 변수 접근을 제공하는 :class:`~airflow.sdk.Variable`, 태스크 간 통신 데이터 접근에 쓰이는 :class:`~airflow.sdk.execution_time.xcom.XCom`.

Connection과 Variable 조작은 :func:`~airflow.sdk.get_current_context`를 통한 Task Context와 태스크 인스턴스의 메서드를 이용하거나, ``airflow.sdk`` 네임스페이스를 통해 수행해야 한다. :class:`~airflow.models.connection.Connection`과 :class:`~airflow.models.variable.Variable` 모델에 대한 직접 데이터베이스 접근은 태스크 코드에서 더 이상 허용되지 않는다.

Task Context를 통한 접근 예: ``context = get_current_context()``; ``conn = context["conn"]``; ``my_connection = conn.get("my_connection_id")``; ``var = context["var"]``; ``my_variable = var.value.get("my_variable_name")``.

airflow.sdk 네임스페이스를 직접 쓰는 예: ``from airflow.sdk import Connection, Variable``; ``conn = Connection.get("my_connection_id")``; ``var = Variable.get("my_variable_name")``.

Connection, Variable, XCom 클래스는 이제 airflow.sdk 네임스페이스의 일부다.

Plugin으로 확장하는 지점 - Trigger, Timetable, Listener, Extra Link

Airflow는 Plugin 메커니즘으로 플랫폼 기능을 확장한다. Plugin은 Airflow UI를 확장할 뿐 아니라, 아래에 나열된 커스터마이징(Trigger, Timetable, Listener 등)을 노출하는 방식이기도 하다. 프로바이더도 플러그인 엔드포인트를 구현해 Airflow UI와 커스터마이징을 확장할 수 있다.

Triggers: Airflow는 ``asyncio`` 호환 Deferrable Operator를 구현하기 위해 Trigger를 사용한다. 모든 Trigger는 :class:`~airflow.triggers.base.BaseTrigger`에서 파생되며, 공개로 취급되는 Trigger 집합이 있어 자유롭게 확장할 수 있다.

Timetables: 커스텀 Timetable 구현은 내장 스케줄 표현식으로는 불가능한 방식으로 Dag Run을 스케줄링하는 추가 로직을 스케줄러에 제공한다. 모든 Timetable은 :class:`~airflow.timetables.base.Timetable`에서 파생되며, 공개로 취급되는 Timetable 집합이 있어 자유롭게 확장할 수 있다.

Listeners: Dag/Task 생애주기 이벤트에 반응할 수 있게 해준다. :class:`~airflow.listeners.listener.ListenerManager` 클래스를 통해 구현되며, Dag/Task 생애주기 이벤트에 반응하는 훅을 구현할 수 있다. Listener 공개 인터페이스는 Airflow 2.5에서 추가되었다.

Extra Links: 커스텀 Operator와 독립적으로 Airflow에 추가될 수 있는 동적 링크다. 보통 Operator가 정의하지만, Plugin을 통해 전역 레벨에서 링크를 오버라이드할 수도 있다.

Executor, Secrets Backend, Auth Manager - 베이스는 공개, 구현체는 컴포넌트마다 다름

Executors는 태스크 인스턴스가 실행되는 메커니즘이다. 모든 executor는 :class:`~airflow.executors.base_executor.BaseExecutor`에서 파생되며, Airflow에는 각기 다른 특성과 능력을 가진 여러 내장 executor 구현체가 있다. executor 인터페이스 자체(BaseExecutor 클래스)는 공개지만, 내장 executor들(KubernetesExecutor, LocalExecutor 등)은 공개가 아니다 - 즉 KubernetesExecutor를 예로 들면, 마이너나 패치 릴리스에서도 KubernetesExecutor에 변경을 가할 수 있고, 이는 KubernetesExecutor를 서브클래싱한 executor를 깨뜨릴 수 있다. 이는 Airflow 개발자가 우리가 제공하는 executor를 계속 개선할 충분한 자유를 갖도록 필요한 조치다. 따라서 내장 executor를 수정·확장하고 싶다면, 그런 변경이 자신의 파생 executor를 깨뜨리지 않도록 전체 executor 코드를 자신의 프로젝트에 통합해야 한다.

.. versionadded:: 2.6

executor 인터페이스 자체는 오래전부터 Airflow에 있었지만, 2.6 이전에는 executor 전용 코드가 코드베이스 여기저기에 흩어져 있었다. 2.6부터 executor는 완전히 분리(decoupled)되어, Airflow 코어가 특정 executor의 동작을 알 필요가 없어졌다. 2.6 이전에도 커스텀 executor 구현에 성공한 사례들은 있었지만, 일부 하드코딩된 동작이 내장 executor를 편애했고 커스텀 executor는 내장 executor가 가진 완전한 기능을 제공할 수 없었다.

Secrets Backends: Airflow는 :class:`~airflow.sdk.Connection`과 :class:`~airflow.sdk.Variable`을 가져오기 위해 secrets backend에 의존하도록 설정될 수 있다. 모든 secrets backend는 :class:`~airflow.secrets.base_secrets.BaseSecretsBackend`에서 파생된다. Executor와 달리, **모든** Secrets Backend 구현체는 공개이며 자유롭게 확장할 수 있다.

Auth managers: 사용자 인증·인가를 담당한다. 모든 auth manager는 :class:`~airflow.api_fastapi.auth.managers.base_auth_manager.BaseAuthManager`에서 파생된다. auth manager 인터페이스 자체(BaseAuthManager 클래스)는 공개지만, auth manager의 서로 다른 구현체들(예: FabAuthManager)은 공개가 아니다.

그 외 확장 지점으로 Connections(커스텀 커넥션 추가), Extra Links(훅에 커스텀 Extra Link 추가), Logging and Monitoring(로그 작성 방식 확장), Decorators(:class:`~airflow.sdk.bases.decorator.TaskDecorator`에서 파생, 주요 데코레이터는 이제 airflow.sdk 네임스페이스 - dag/task/asset/setup/task_group/teardown/chain/chain_linear/cross_downstream/get_current_context/get_parsing_context), Email notifications, ``on_*_callback`` 기반 Notifications, Cluster Policies(파싱/실행 중인 Dag·태스크에 클러스터 전역 정책을 동적으로 적용), Lineage(데이터의 출처와 이동 경로 추적)가 있다.