리스너

리스너 (Listeners)

이 페이지는 Airflow에서 이벤트가 발생했을 때 알림을 받도록 하는 리스너(Listeners) 작성법을 다뤄요. 리스너는 Pluggy로 구현되며, Airflow Job의 시작·종료, DagRun 상태 변경, TaskInstance 상태 변경, Asset 이벤트, DAG import 에러 이벤트 등 여러 이벤트를 구독할 수 있어요. 버전별 인터페이스 호환성에 주의해야 해요.

출처: 문서

본문

리스너를 작성하면 이벤트가 발생했을 때 Airflow가 알림을 보내도록 할 수 있어요. 이 리스너는 Pluggy로 구동돼요.

Warning

리스너는 Airflow의 고급 기능이에요. 리스너는 그것이 실행되는 Airflow 컴포넌트와 격리되지 않으며, Airflow 인스턴스를 느리게 하거나 어떤 경우에는 다운시킬 수도 있어요. 따라서 리스너를 작성할 때 특히 주의해야 해요.

Airflow는 다음 이벤트에 대한 알림을 지원해요:

라이프사이클 이벤트

• on_starting
• before_stopping

라이프사이클 이벤트를 사용하면 SchedulerJob 같은 Airflow Job의 시작·종료 이벤트에 반응할 수 있어요.

DagRun 상태 변경 이벤트

DagRun 상태 변경 이벤트는 DagRun이 상태를 변경할 때 발생해요. Airflow 3부터는 API를 통해 상태 변경이 트리거될 때도 리스너가 알림을 받아요(on_dag_run_success와 on_dag_run_failed의 경우) — 예를 들어 Airflow UI에서 DagRun을 success로 표시할 때요.

• on_dag_run_running

airflow/example_dags/plugins/event_listener.py[source]

@hookimpl
def on_dag_run_running(dag_run: DagRun, msg: str):
    """
    This method is called when dag run state changes to RUNNING.
    """
    print("Dag run  in running state")
    queued_at = dag_run.queued_at

    version = dag_run.version_number

    print(f"Dag information Queued at: {queued_at} version: {version}")

• on_dag_run_success

airflow/example_dags/plugins/event_listener.py[source]

@hookimpl
def on_dag_run_success(dag_run: DagRun, msg: str):
    """
    This method is called when dag run state changes to SUCCESS.
    """
    print("Dag run in success state")
    start_date = dag_run.start_date
    end_date = dag_run.end_date

    print(f"Dag run start:{start_date} end:{end_date}")

• on_dag_run_failed

airflow/example_dags/plugins/event_listener.py[source]

@hookimpl
def on_dag_run_failed(dag_run: DagRun, msg: str):
    """
    This method is called when dag run state changes to FAILED.
    """
    print("Dag run  in failure state")
    dag_id = dag_run.dag_id
    run_id = dag_run.run_id
    run_type = dag_run.run_type

    print(f"Dag information:{dag_id} Run id: {run_id} Run type: {run_type}")
    print(f"Failed with message: {msg}")

TaskInstance 상태 변경 이벤트

TaskInstance 상태 변경 이벤트는 RuntimeTaskInstance가 상태를 변경할 때 발생해요. 이 이벤트를 사용해 LocalTaskJob 상태 변경에 반응할 수 있어요. Airflow 3부터는 API를 통해 상태 변경이 트리거될 때도 리스너가 알림을 받아요(on_task_instance_success와 on_task_instance_failed의 경우) — 예를 들어 Airflow UI에서 task instance를 success로 표시할 때요. 그런 경우 리스너는 RuntimeTaskInstance 인스턴스 대신 TaskInstance 인스턴스를 받아요.

• on_task_instance_running

airflow/example_dags/plugins/event_listener.py[source]

@hookimpl
def on_task_instance_running(
    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
):
    """
    Called when task state changes to RUNNING.

    previous_task_state and task_instance object can be used to retrieve more information about current
    task_instance that is running, its dag_run, task and dag information.
    """
    print("Task instance is in running state")
    print(" Previous state of the Task instance:", previous_state)

    name: str = task_instance.task_id

    context = task_instance.get_template_context()

    task = context["task"]

    if TYPE_CHECKING:
        assert task

    dag = task.dag
    dag_name = None
    if dag:
        dag_name = dag.dag_id
    print(f"Current task name:{name}")
    print(f"Dag name:{dag_name}")

• on_task_instance_success

airflow/example_dags/plugins/event_listener.py[source]

@hookimpl
def on_task_instance_success(
    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
):
    """
    Called when task state changes to SUCCESS.

    previous_task_state and task_instance object can be used to retrieve more information about current
    task_instance that has succeeded, its dag_run, task and dag information.

    A RuntimeTaskInstance is provided in most cases, except when the task's state change is triggered
    through the API. In that case, the TaskInstance available on the API server will be provided instead.
    """
    print("Task instance in success state")
    print(" Previous state of the Task instance:", previous_state)

    if isinstance(task_instance, TaskInstance):
        print("Task instance's state was changed through the API.")

        print(f"Task operator:{task_instance.operator}")
        return

    context = task_instance.get_template_context()
    operator = context["task"]

    print(f"Task operator:{operator}")

• on_task_instance_failed

airflow/example_dags/plugins/event_listener.py[source]

@hookimpl
def on_task_instance_failed(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
    error: None | str | BaseException,
):
    """
    Called when task state changes to FAILED.

    previous_task_state, task_instance object and error can be used to retrieve more information about current
    task_instance that has failed, its dag_run, task and dag information.

    A RuntimeTaskInstance is provided in most cases, except when the task's state change is triggered
    through the API. In that case, the TaskInstance available on the API server will be provided instead.
    """
    print("Task instance in failure state")

    if isinstance(task_instance, TaskInstance):
        print("Task instance's state was changed through the API.")

        print(f"Task operator:{task_instance.operator}")
        if error:
            print(f"Failure caused by {error}")
        return

    context = task_instance.get_template_context()
    task = context["task"]

    if TYPE_CHECKING:
        assert task

    print("Task start")
    print(f"Task:{task}")
    if error:
        print(f"Failure caused by {error}")

• on_task_instance_skipped

airflow/example_dags/plugins/event_listener.py[source]

@hookimpl
def on_task_instance_skipped(
    previous_state: TaskInstanceState | None, task_instance: RuntimeTaskInstance | TaskInstance
):
    """
    Called when a task instance skips itself during execution.

    This hook is called only when a task has started execution and then
    intentionally skips itself (e.g., by raising AirflowSkipException).

    Note: This function will NOT cover tasks that were skipped by scheduler, before execution began, such as:
        - Skips due to trigger rules (e.g., upstream failures)
        - Skips from operators like BranchPythonOperator, ShortCircuitOperator, or similar mechanisms
        - Any other situation in which the scheduler decides not to schedule a task for execution

    For comprehensive tracking of skipped tasks, use DAG-level listeners
    (on_dag_run_success/on_dag_run_failed) which may have access to all task states.
    """
    print("Task instance was skipped")

    if isinstance(task_instance, TaskInstance):
        print("Task instance's state was changed through the API.")
        return

    context = task_instance.get_template_context()
    task = context["task"]

    if TYPE_CHECKING:
        assert task

    print("Task start")
    print(f"Task:{task}")

Asset 이벤트

• on_asset_created
• on_asset_alias_created
• on_asset_changed

Asset 이벤트는 Asset 관리 작업이 실행될 때 발생해요.

DAG import 에러 이벤트

• on_new_dag_import_error
• on_existing_dag_import_error

DAG import 에러 이벤트는 Dag processor가 DAG 코드에서 import 에러를 발견하고 메타데이터 데이터베이스 테이블을 업데이트할 때 발생해요.

이것은 실험적 기능이에요.

사용법

리스너를 만들려면:

• airflow.listeners.hookimpl을 import해요.
• 알림을 생성하고 싶은 이벤트에 대해 hookimpl를 구현해요.

Airflow는 사양(specification)을 hookspec으로 정의해요. 구현은 hookspec에 정의된 것과 같은 명명된 파라미터를 받아들여야 해요. hookspec과 같은 파라미터를 사용하지 않으면 플러그인을 사용하려 할 때 Pluggy가 에러를 던져요. 하지만 모든 메서드를 구현할 필요는 없어요. 많은 리스너는 하나의 메서드 또는 메서드 부분집합만 구현해요.

Airflow 설치에 리스너를 포함하려면 Airflow 플러그인의 일부로 포함시켜요.

Listener API는 모든 DAG와 모든 Operator에 걸쳐 호출되도록 의도되었어요. 특정 DAG에서 생성된 이벤트만 들을 수는 없어요. 그런 동작이 필요하면 on_success_callback과 pre_execute 같은 메서드를 사용해 보세요. 이들은 특정 DAG 작성자나 Operator 생성자를 위한 콜백을 제공해요. 로그와 print() 호출은 리스너의 일부로 처리돼요.

호환성 참고

리스너 인터페이스는 시간이 지나면서 바뀔 수 있어요. 우리는 pluggy 사양을 사용하므로, 인터페이스의 구버전을 위해 작성된 리스너 구현은 미래 버전의 Airflow와 정방향 호환(forward-compatible)이어야 해요.

하지만 그 반대는 보장되지 않아요. 새 버전의 인터페이스에 맞춰 구현된 리스너는 Airflow 구버전에서 작동하지 않을 수 있어요. 단일 버전의 Airflow를 대상으로 한다면 문제가 없어요(사용하는 Airflow 버전에 맞춰 구현을 조정하면 되니까요). 하지만 다른 버전의 Airflow에서 사용될 수 있는 플러그인이나 확장을 작성한다면 중요해요.

예를 들어 인터페이스에 새 필드가 추가되면(2.10.0의 on_task_instance_failed 메서드의 error 필드처럼), 리스너 구현은 이벤트 객체에 필드가 없을 때를 처리하지 못하며, 그런 리스너는 Airflow 2.10.0 이상에서만 작동해요.

새 버전의 Airflow에서 추가된 기능과 필드를 사용하면서 여러 버전의 Airflow와 호환되는 리스너를 구현하려면, 사용 중인 Airflow 버전을 확인해 Airflow의 새 버전에서는 새 인터페이스 구현을, 구버전에서는 구버전 인터페이스 구현을 사용해야 해요.

예를 들어 on_task_instance_failed에서 error 필드를 사용하는 리스너를 구현하려면 다음과 같은 코드를 사용해요:

from importlib.metadata import version
from packaging.version import Version
from airflow.listeners import hookimpl

airflow_version = Version(version("apache-airflow"))
if airflow_version >= Version("2.10.0"):

    class ClassBasedListener:
        ...

        @hookimpl
        def on_task_instance_failed(self, previous_state, task_instance, error: None | str | BaseException):
            # Handle error case here
            pass

else:

    class ClassBasedListener:  # type: ignore[no-redef]
        ...

        @hookimpl
        def on_task_instance_failed(self, previous_state, task_instance):
            # Handle no error case here
            pass

리스너 인터페이스가 도입된 2.8.0 이후 인터페이스의 변경 목록:

Airflow 버전	영향받는 메서드	변경 사항
2.10.0	on_task_instance_failed	인터페이스에 error 필드 추가
3.0.0	on_task_instance_running	task instance 리스너에서 session 인자 제거, task_instance 객체가 이제 RuntimeTaskInstance 인스턴스
3.0.0	on_task_instance_failed, on_task_instance_success	task instance 리스너에서 session 인자 제거, task_instance 객체가 worker에서는 RuntimeTaskInstance, API server에서는 TaskInstance 인스턴스
3.2.0	on_task_instance_skipped	인터페이스에 새 리스너 메서드 추가

더 알아보기 (Learn more)

• Plugins
• Lineage
• Cluster Policies
• Callbacks