플러그인

플러그인 (Plugins)

이 페이지는 Airflow의 플러그인 시스템을 다뤄요. $AIRFLOW_HOME/plugins 폴더에 파일을 넣기만 하면 외부 기능을 Airflow 코어에 통합할 수 있어요. Airflow 3.1부터는 React 앱, FastAPI 엔드포인트, 미들웨어 같은 새 기능도 지원해요. 플러그인이 등록할 수 있는 빌딩 블록(External Views, Macros, Operator Extra Links 등)과 작성 예제를 설명해요.

출처: 문서

본문

Airflow에는 $AIRFLOW_HOME/plugins 폴더에 파일을 넣기만 하면 외부 기능을 코어에 통합할 수 있는 간단한 플러그인 매니저가 내장되어 있어요.

Airflow 3.1부터 플러그인 시스템은 React 앱, FastAPI 엔드포인트, 미들웨어 같은 새 기능을 지원해서, Airflow를 확장하고 풍부한 커스텀 통합을 구축하기가 더 쉬워졌어요.

plugins 폴더의 Python 모듈은 import되며, **매크로(macros)**와 웹 **뷰(views)**가 Airflow의 주요 컬렉션에 통합되어 사용할 수 있게 돼요.

플러그인 문제를 해결하려면 airflow plugins 명령을 사용할 수 있어요. 이 명령은 로드된 플러그인에 대한 정보를 출력해요.

왜 사용하나요?

Airflow는 데이터 작업을 위한 일반적인 도구 상자를 제공해요. 조직마다 스택과 요구사항이 달라요. Airflow 플러그인을 사용하면 회사가 자신의 생태계를 반영하도록 Airflow 설치를 커스터마이즈할 수 있어요.

플러그인은 새 기능 세트를 작성·공유·활성화하는 쉬운 방법으로 사용할 수 있어요.

또한 다양한 형태의 데이터와 메타데이터와 상호작용하는 더 복잡한 애플리케이션 세트에 대한 필요도 있어요.

예시:

• Hive 로그를 파싱하고 Hive 메타데이터(CPU/IO/단계/skew/…)를 노출하는 도구 세트
• 메트릭을 수집하고 임계값과 알림을 설정할 수 있는 이상 탐지(anomaly detection) 프레임워크
• 누가 무엇에 접근하는지 이해하는 데 도움을 주는 감사(auditing) 도구
• 모니터링할 테이블과 그것이 언제 도착해야 하는지를 설정하고, 사람들에게 알리며, 중단 시각화를 노출하는 설정 기반 SLA 모니터링 도구

왜 Airflow 위에 구축하나요?

Airflow에는 애플리케이션을 구축할 때 재사용할 수 있는 많은 컴포넌트가 있어요:

• 뷰를 렌더링하는 데 사용할 수 있는 웹 서버
• 모델을 저장할 메타데이터 데이터베이스
• 데이터베이스에 대한 접근과 연결 방법에 대한 지식
• 애플리케이션이 워크로드를 밀어 넣을 수 있는 worker 배열
• Airflow는 배포되어 있으므로, 그 배포 로지스틱에 편승(piggyback)할 수 있어요
• 기본 차트 기능, 기반 라이브러리와 추상화

사용 가능한 빌딩 블록

Airflow 플러그인은 다음 컴포넌트를 등록할 수 있어요:

• External Views – UI에 새 페이지로 연결하는 버튼/탭을 추가해요.
• React Apps – Airflow UI 안에 커스텀 React 앱을 내장해요 (Airflow 3.1에서 새로 추가).
• FastAPI Apps – 커스텀 API 엔드포인트를 추가해요.
• FastAPI Middlewares – API 요청/응답을 가로채고 수정해요.
• Macros – DAG 템플릿에서 사용 가능한 재사용 가능한 Python 함수를 정의해요.
• Operator Extra Links – Task 상세 뷰에 커스텀 버튼을 추가해요.
• Timetables & Listeners – 커스텀 스케줄링 로직과 이벤트 훅을 구현해요.

플러그인은 언제 (재)로딩되나요?

플러그인은 기본적으로 지연 로드(lazily loaded)되며, 한 번 로드되면 다시 로드되지 않아요(UI 플러그인은 Webserver에서 자동으로 로드되는 점 제외). 각 Airflow 프로세스 시작 시 로드하려면 airflow.cfg의 [core] lazy_load_plugins = False를 설정해요.

즉 플러그인을 변경했고 webserver나 scheduler가 그 새 코드를 사용하길 원한다면, 해당 프로세스를 재시작해야 해요. 하지만 새로 실행되는 Task에는 scheduler가 부팅된 후에야 반영돼요.

기본적으로 Task 실행은 forking을 사용해요. 이는 새 Python 인터프리터를 만들고 Airflow의 모든 코드와 시작 루틴을 다시 파싱하는 것과 관련된 속도 저하를 피해요. 이 접근 방식은 특히 짧은 Task에서 상당한 이점을 제공해요. 이는 Task에서 플러그인을 사용하고 업데이트하길 원한다면, worker(CeleryExecutor 사용 시)나 scheduler(LocalExecutor 사용 시)를 재시작해야 한다는 뜻이에요. 다른 옵션으로 시작 시 속도 저하를 감수하고 core.execute_tasks_new_python_interpreter 설정을 True로 해서 Task를 위해 완전히 새로운 Python 인터프리터를 띄울 수 있어요.

(반면 Dag 파일에서만 import되는 모듈은 이 문제를 겪지 않아요. Dag 파일은 장기 실행되는 Airflow 프로세스에서 로드/파싱되지 않기 때문이에요.)

인터페이스

플러그인을 만들려면 airflow.plugins_manager.AirflowPlugin 클래스를 파생하고 Airflow에 연결하고 싶은 객체를 참조해야 해요. 파생해야 하는 클래스는 이렇게 생겼어요:

class AirflowPlugin:
    # The name of your plugin (str)
    name = None
    # A list of references to inject into the macros namespace
    macros = []
    # A list of dictionaries containing FastAPI app objects and some metadata. See the example below.
    fastapi_apps = []
    # A list of dictionaries containing FastAPI middleware factory objects and some metadata. See the example below.
    fastapi_root_middlewares = []
    # A list of dictionaries containing external views and some metadata. See the example below.
    external_views = []
    # A list of dictionaries containing react apps and some metadata. See the example below.
    # Note: React apps are only supported in Airflow 3.1 and later.
    # Note: The React app integration is experimental and interfaces might change in future versions. Particularly, dependency and state interactions between the UI and plugins may need to be refactored for more complex plugin apps.
    react_apps = []

    # A callback to perform actions when Airflow starts and the plugin is loaded.
    # NOTE: Ensure your plugin has *args, and **kwargs in the method definition
    #   to protect against extra parameters injected into the on_load(...)
    #   function in future changes
    def on_load(*args, **kwargs):
        # ... perform Plugin boot actions
        pass

    # A list of global operator extra links that can redirect users to
    # external systems. These extra links will be available on the
    # task page in the form of buttons.
    #
    # Note: the global operator extra link can be overridden at each
    # operator level.
    global_operator_extra_links = []

    # A list of operator extra links to override or add operator links
    # to existing Airflow Operators.
    # These extra links will be available on the task page in form of
    # buttons.
    operator_extra_links = []

    # A list of timetable classes to register so they can be used in Dags.
    timetables = []

    # A list of deadline reference classes that can be used as custom deadlines in Dags.
    # Custom deadline reference classes must be registered here in order to be
    # resolvable at scheduler-side deserialization time; classes that are not
    # registered will raise ``DeadlineReferenceNotRegistered`` when a Dag attempts
    # to use them.
    deadline_references = []

    # A list of Listeners that plugin provides. Listeners can register to
    # listen to particular events that happen in Airflow, like
    # TaskInstance state changes. Listeners are python modules.
    listeners = []

상속으로 파생할 수 있어요(아래 예제 참고). 예제에서 모든 옵션은 클래스 속성으로 정의됐지만, 추가 초기화를 수행해야 한다면 속성(property)으로 정의할 수도 있어요. 이 클래스 안의 name은 반드시 지정해야 해요.

플러그인 변경 후에는 적용되도록 webserver와 scheduler를 재시작해야 해요.

플러그인 관리 인터페이스

Airflow 3.1은 Airflow UI의 Admin → Plugins 아래에 플러그인 관리 인터페이스를 도입해요. 이 페이지에서 설치된 플러그인을 볼 수 있어요.

External Views

External views는 url_route 값을 제공해 iframe을 사용해 Airflow UI에 직접 내장할 수도 있어요. 이렇게 하면 새 브라우저 탭에서 여는 대신 뷰를 인라인으로 렌더링할 수 있어요.

예제

아래 코드는 Airflow에 일련의 설명용 객체 정의를 주입하는 플러그인을 정의해요.

# This is the class you derive to create a plugin
from airflow.plugins_manager import AirflowPlugin

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware

# Importing base classes that we need to derive
from airflow.hooks.base import BaseHook
from airflow.providers.amazon.aws.transfers.gcs_to_s3 import GCSToS3Operator

# Will show up in templates through {{ macros.test_plugin.plugin_macro }}
def plugin_macro():
    pass

# Creating a FastAPI application to integrate in Airflow Rest API.
app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World from FastAPI plugin"}

app_with_metadata = {"app": app, "url_prefix": "/some_prefix", "name": "Name of the App"}

Warning

Airflow는 플러그인 FastAPI 앱을 인증하지 않아요. 이를 인증하는 것은 플러그인 작성자의 책임이에요.

Airflow는 router 레벨에서, 일부 엔드포인트는 라우트별로 선언된 인증 의존성으로 핵심 API를 인증해요. 플러그인 앱은 app.mount()로 연결되는데, Starlette mount는 자체 라우트 테이블을 가지며 부모의 의존성을 전혀 상속하지 않으므로, 그 의존성은 플러그인의 라우트에 도달하지 않아요. API 서버의 어떤 미들웨어도 그들을 인증하지 않아요.

따라서 플러그인이 노출하는 모든 라우트는 플러그인이 스스로 인증하지 않는 한 익명 호출자에게 도달 가능해요. 위의 최소 app은 구조적 설명일 뿐, 그대로 배포할 템플릿이 아니에요.

Airflow가 인증한 호출자를 요구하려면 GetUserDep에 의존하세요:

from fastapi import FastAPI

from airflow.api_fastapi.core_api.security import GetUserDep

app = FastAPI()

@app.get("/dashboard")
def dashboard(user: GetUserDep):
    return {"user": user.get_name()}

애플리케이션이나 router 레벨에서 의존성을 한 번 붙이는 것을 선호하세요. 그래야 나중에 추가된 라우트가 조용히 인증되지 않은 채 배포되지 않아요:

from fastapi import Depends, FastAPI

from airflow.api_fastapi.core_api.security import get_user

app = FastAPI(dependencies=[Depends(get_user)])

인증은 권한 부여(authorization)가 아니에요. GetUserDep는 누가 호출하는지 확립해요. 그 사용자가 주어진 작업을 수행할 수 있는지는 플러그인 자신의 결정으로 남아요. 이는 팀 스코핑에도 적용돼요 — 멀티 팀 배포에서 호출자의 팀을 확인하지 않는 플러그인은 모든 팀 사용자에게 같은 데이터를 제공해요.

핵심 API의 접근 헬퍼가 그 결정을 강제할 수 있어요. 예를 들어 requires_access_dag는 라우트를 DAG에 대해 요청된 작업이 허용된 호출자로 제한해요. 호출자를 인증하고 요청에서 dag_id를 읽어요:

from fastapi import Depends, FastAPI

from airflow.api_fastapi.core_api.security import requires_access_dag

app = FastAPI()

@app.get("/dags/{dag_id}", dependencies=[Depends(requires_access_dag(method="GET"))])
def dag_detail(dag_id: str):
    return {"dag_id": dag_id}

# Creating a FastAPI middleware that will operates on all the server api requests.
middleware_with_metadata = {
    "middleware": TrustedHostMiddleware,
    "args": [],
    "kwargs": {"allowed_hosts": ["example.com", "*.example.com"]},
    "name": "Name of the Middleware",
}

# Creating an external view that will be rendered in the Airflow UI.
external_view_with_metadata = {
    # Name of the external view, this will be displayed in the UI.
    "name": "Name of the External View",
    # Source URL of the external view. This URL can be templated using context variables, depending on the location where the external view is rendered
    # the context variables available will be different, i.e a subset of (DAG_ID, RUN_ID, TASK_ID, MAP_INDEX).
    "href": "https://example.com/{DAG_ID}/{RUN_ID}/{TASK_ID}/{MAP_INDEX}",
    # Destination of the external view. This is used to determine where the view will be loaded in the UI.
    # Supported locations are Literal["nav", "dag", "dag_run", "task", "task_instance", "base"], default to "nav".
    "destination": "dag_run",
    # Optional icon, url to an svg file.
    "icon": "https://example.com/icon.svg",
    # Optional dark icon for the dark theme, url to an svg file. If not provided, "icon" will be used for both light and dark themes.
    "icon_dark_mode": "https://example.com/dark_icon.svg",
    # Optional parameters, relative URL location for the External View rendering. If not provided, external view will be rendered as an external link. If provided
    # will be rendered inside an Iframe in the UI. Should not contain a leading slash.
    "url_route": "my_external_view",
    # Optional category, only relevant for destination "nav". This is used to group the external links in the navigation bar.  We will match the existing
    # menus of ["browse", "docs", "admin", "user"] and if there's no match then create a new menu.
    "category": "browse",
    # Optional flag, only relevant for destination "nav". When True, this item is always rendered directly on the
    # navigation toolbar instead of inside the "Plugins" submenu. When two or more non-promoted items remain they
    # are still grouped into the submenu; a single remaining non-promoted item is also shown on the toolbar.
    # Defaults to False.
    "nav_top_level": True,
}

# Note: The React app integration is experimental and interfaces might change in future versions.
react_app_with_metadata = {
    # Name of the React app, this will be displayed in the UI.
    "name": "Name of the React App",
    # Bundle URL of the React app. This is the URL where the React app is served from. It can be a static file or a CDN.
    # This URL can be templated using context variables, depending on the location where the external view is rendered
    # the context variables available will be different, i.e a subset of (DAG_ID, RUN_ID, TASK_ID, MAP_INDEX).
    "bundle_url": "https://example.com/static/js/my_react_app.js",
    # Destination of the react app. This is used to determine where the app will be loaded in the UI.
    # Supported locations are Literal["nav", "dag", "dag_run", "task", "task_instance", "base"], default to "nav".
    # It can also be put inside of an existing page, the supported views are ["dashboard", "dag_overview", "task_overview"]. You can position
    # element in the existing page via the css `order` rule which will determine the flex order.
    # Use "base" to mount the app in the base layout (e.g. a toolbar strip); the host uses a flex container so you can set ``order`` in your root JSX to control position.
    "destination": "dag_run",
    # Optional icon, url to an svg file.
    "icon": "https://example.com/icon.svg",
    # Optional dark icon for the dark theme, url to an svg file. If not provided, "icon" will be used for both light and dark themes.
    "icon_dark_mode": "https://example.com/dark_icon.svg",
    # URL route for the React app, relative to the Airflow UI base URL. Should not contain a leading slash.
    "url_route": "my_react_app",
    # Optional category, only relevant for destination "nav". This is used to group the react apps in the navigation bar. We will match the existing
    # menus of ["browse", "docs", "admin", "user"] and if there's no match then create a new menu.
    "category": "browse",
    # Optional flag, only relevant for destination "nav". When True, this item is always rendered directly on the
    # navigation toolbar instead of inside the "Plugins" submenu. When two or more non-promoted items remain they
    # are still grouped into the submenu; a single remaining non-promoted item is also shown on the toolbar.
    # Defaults to False.
    "nav_top_level": True,
}

# Defining the plugin class
class AirflowTestPlugin(AirflowPlugin):
    name = "test_plugin"
    macros = [plugin_macro]
    fastapi_apps = [app_with_metadata]
    fastapi_root_middlewares = [middleware_with_metadata]
    external_views = [external_view_with_metadata]
    react_apps = [react_app_with_metadata]

See also

Define an operator extra link

CSRF 보호에서 뷰 제외하기

모든 뷰를 CSRF로 보호할 것을 강력히 권장해요. 하지만 필요하다면 데코레이터를 사용해 일부 뷰를 제외할 수 있어요.

from airflow.www.app import csrf

@csrf.exempt
def my_handler():
    # ...
    return "ok"

Python 패키지로서의 플러그인

setuptools entrypoint 매커니즘을 통해 플러그인을 로드하는 것도 가능해요. 그러려면 패키지에서 entrypoint를 사용해 플러그인을 연결해요. 패키지가 설치되면 Airflow는 entrypoint 목록에서 등록된 플러그인을 자동으로 로드해요.

Note

entrypoint 이름(예: my_plugin)도, 플러그인 클래스 이름도 플러그인 자체의 모듈·클래스 이름에 기여하지 않아요.

# my_package/my_plugin.py
from airflow.plugins_manager import AirflowPlugin

class MyAirflowPlugin(AirflowPlugin):
    name = "my_namespace"

그다음 pyproject.toml 안에:

[project.entry-points."airflow.plugins"]
my_plugin = "my_package.my_plugin:MyAirflowPlugin"

Airflow 3의 Flask Appbuilder와 Flask Blueprints

Airflow 2는 플러그인에서 Flask Appbuilder 뷰(appbuilder_views), Flask AppBuilder 메뉴 항목(appbuilder_menu_items), Flask Blueprints(flask_blueprints)를 지원했어요. 이것들은 Airflow 3에서 External Views(external_views), Fast API apps(fastapi_apps), FastAPI middlewares(fastapi_root_middlewares), React apps(react_apps)으로 대체되어, 확장된 기능과 Airflow UI와의 더 나은 통합을 제공해요.

모든 새 플러그인은 새 인터페이스를 사용해야 해요.

하지만 Airflow 3로의 전환을 쉽게 하기 위해 Flask와 FAB 플러그인용 호환성 레이어가 제공돼요 — FAB provider를 설치하고 Airflow 3 마이그레이션 가이드에 따라 코드를 튜닝하면 돼요. 이 호환성 레이어를 사용하면 기존의 Flask Appbuilder 뷰, Flask Blueprints, Flask Appbuilder 메뉴 항목을 계속 사용할 수 있어요.

문제 해결

Flask CLI를 사용해 문제를 해결할 수 있어요. 실행하려면 FLASK_APP 변수를 airflow.www.app:create_app으로 설정해야 해요.

예를 들어 모든 라우트를 출력하려면 다음을 실행해요:

FLASK_APP=airflow.www.app:create_app flask routes

더 알아보기 (Learn more)

• Modules Management
• Listeners
• Lineage
• Define an operator extra link