5. Task, Operator, Sensor, Hook, Provider

알아볼 내용

• Task 정의, 상태
• Task 간 순서 (Dependency) 정의
• XCom, Connection
• Operator
• Sensor, Deferrable Operator
• Hook
• Provider, Plugin

---

1. Task란?

Task는 Airflow에서 실행되는 작업의 기본 단위로, DAG의 노드로 배치되고 Dependency 설정을 통해 실행 순서를 결정할 수 있다

Task에는 세 가지 기본 유형이 존재한다

• Operator
• Sensor (Operator의 일종)
• @task 로 만든 함수 Task

내부적으로는 모두 BaseOperator의 서브클래스이며 개념상 Task는 실행 인스턴스, Operator는 재사용 가능한 템플릿을 의미한다

Task 상태

DAG를 실행하면 정의된 대로 Task가 실행되는데 이때 Task는 다양한 상태를 가질 수 있다
Airflow UI에서 Task의 현재 상태를 쉽게 파악하는데 도움이 된다

• none: 작업이 아직 실행 대기열에 들어가지 않은 상태 (의존성이 아직 충족되지 않음)
• scheduled: Scheduler가 작업의 의존성이 모두 충족되었음을 확인하고, 실행해야 한다고 결정한 상태
• queued: 작업이 Executor에 할당되어, Worker를 기다리고 있는 상태
• running: 작업이 Worker (또는 로컬/동기식 executor)에서 실행 중인 상태
• success: 작업이 오류 없이 정상적으로 완료된 상태
• restarting: 작업이 실행 중일 때 외부에서 재시작 요청을 받은 상태
• failed: 작업 실행 중 오류가 발생하여 실패한 상태
• skipped: branching, LatestOnly 등으로 인해 작업이 건너뛰어진 상태
• upstream_failed: Upstream 작업이 실패했고,  Trigger Rules에 따라 이 작업이 필요했던 경우
• up_for_retry: 작업이 실패했지만, 재시도 횟수가 남아 있어 다시 실행될 예정인 상태
• up_for_reschedule: 작업이 Sensor 이고, reschedule 모드에서 다시 예약 대기 중인 상태
• deferred: 작업이 trigger로 연기된 상태 (Deferred)
• removed: DAG 실행이 시작된 이후 작업이 DAG에서 사라진 상태

Task Dependency

Airflow에서는 Task 간 의존성을 직관적으로 정의할 수 있다
의존성을 잘 설정해야 DAG이 올바른 순서로 실행되고, 병렬/분기 실행도 원활히 관리된다.

1. 기본 연결 연산자

• >> : 순방향

  task1 >> task2   # task1 실행 후 task2 실행
  task1.set_downstream(task2) # 동일한 표현

• << : 역방향

  task2 << task1   # task1 실행 후 task2 실행
  task2.set_upstream(task1)  # 동일한 표현

2. 다중 Task 연결

• 여러 태스크가 모여서 하나로 합쳐지는 구조

  [task1, task2] >> task3   # task1, task2가 모두 성공해야 task3 실행 (Trigger Rules 참고)

• 반대로 하나의 태스크에서 여러 개로 분기

  task1 >> [task2, task3]

3. Cross Downstream

• 두 리스트 간의 모든 조합을 연결

  from airflow.sdk import cross_downstream
   
  cross_downstream([task1, task2], [task3, task4])
   
  # 결과
  # task1 >> task3
  # task1 >> task4
  # task2 >> task3
  # task2 >> task4

4. Chain

• 순차 체인을 간결하게 표현

  from airflow.sdk import chain
  chain(task1, task2, task3, task4)
   
  # 결과
  # task1 >> task2 >> task3 >> task4

예시

task0 >> task1
[task1, task2] >> task3
task3 >> [task4, task5]
cross_downstream([task4, task5], [task6, task7])
chain(task6, task9, task10)
chain(task7, task8)

XCOM : Task간의 정보교환

Task는 기본적으로 데이터를 주고받지 않는데 이전 Task에서 얻은 결과를 다음 Task에서 받아서 사용하고 싶을 수 있다

XCom(Cross-Communication)은 Airflow에서 태스크 간 작은 데이터 조각을 주고받는 메커니즘으로 데이터를 key-value 형태로 metadata DB (XCom 테이블)에 저장할 수 있다 일반적으로 짧은 문자열·숫자·경로·작은 JSON을 전달하는 용도로 사용한다

기본 동작은 다음과 같다

• ti.xcom_push(key, value) 로 저장, ti.xcom_pull(task_ids, key) 로 조회  
• @task로 정의한 함수의 Return값은 자동으로 XCom에 저장
• 기본 직렬화는 JSON 호환 타입 권장 (복잡 객체는 커스텀 직렬화/백엔드 고려)
• XCom은 TaskInstance 단위로 기록된다 (같은 DAG Run, 같은 태스크 실행 범위)

가장 기본적인 예시는 다음과 같다

import pendulum
from airflow import DAG
from airflow.operators.python import PythonOperator
 
def producer(**context):
    # 어떤 계산/조회 결과라고 가정
    result = {"n": 7, "msg": "hello xcom"}
    context["ti"].xcom_push(key="payload", value=result)
 
def consumer(**context):
    payload = context["ti"].xcom_pull(task_ids="produce", key="payload")
    print("pulled:", payload)  # {'n': 7, 'msg': 'hello xcom'}
 
with DAG(
    dag_id="ex_xcom_push_pull",
    start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
    schedule="@daily",
    catchup=False,
    tags=["xcom", "basic"],
) as dag:
    produce = PythonOperator(task_id="produce", python_callable=producer)
    consume = PythonOperator(task_id="consume", python_callable=consumer)
    
    produce >> consume

XCom을 사용할 때 등장하는 Context와 'ti'?

• Context : Scheduler가 Task를 실행할 때 넘겨주는 metadata dictionary
  • PythonOperator의 python_callable에는 기본적으로 **context가 전달
  • @task로 정의한 함수 안에서도 get_current_context()로 동일한 내용 확인 가능
  • context에 포함된 key 예시
    • ‘ti’ :  TaskInstance 객체 (alias: task_instance)
    • ‘task’, ‘dag’ : 현재 Task, DAG 객체
    • ‘dag_run’ : 현재 DAG Run 객체
    • ‘ds’, ‘ts’ : 논리 실행일의 날짜 문자열(YYYY-MM-DD), 타임스탬프(ISO)
    • ‘logical_date’: pendulum.DateTime 객체
    • ‘data_interval_start’, ‘data_interval_end’: 스케줄 주기의 시작/끝 시각
    • ‘params’: DAG/Task에 주입한 사용자 파라미터 딕셔너리
    • ‘var’ : 템플릿에서 변수 접근용 매크로 (파이썬 코드에서는 Variable.get() 사용 권장)
• ‘ti’ : 현재 실행 중인 태스크의 인스턴스를 표현하는 객체
  • ti.task_id, ti_dag_id, ti.try_number 등 다양한 속성 확인 가능
  • xcom_push로 넣은 값은 metaDB에 저장되기 때문에 context 출력으로 확인안됨

---

2. Control Flow

기본적으로 Task는 모든 upstream task가 성공해야 실행되지만 다음과 같은 방법들을 통해 더 유연한 제어가 가능하다

• Branching
• Trigger Rules
• Latest Only
• Depends On Past
• Setup and Teardown

Branching

만약 어떤 task의 성공 여부에 따라 다음 작업을 다르게 가져가고 싶을 때 Branching을 사용하면 된다

@task.branch() decorator를 사용하거나 BranchPythonOperator를 사용해서 branching 역할을 담당하는 task를 파이썬 함수로 작성하는데 이때 return 값은 다음에 실행할 task_id 를 넣으면 된다

다음은 BranchPythonOperator를 사용하는 예시이다 (decorator 사용은 docs 참고)

from airflow import DAG
from airflow.operators.bash import BashOperatofrom airflow import DAGr
from airflow.operators.python import BranchPythonOperator
 
# DAG 정의
dag = DAG(
    "branch_random_dag",
    default_args=default_args,
    description="0~10 랜덤 값에 따른 브랜치 DAG",
    schedule="@daily",
    catchup=False,
)
 
 
def generate_random_and_branch():
    """0~10 사이의 랜덤 값을 생성하고 5 이상/미만에 따라 브랜치 결정"""
    random_value = random.randint(0, 10)
    print(f"생성된 랜덤 값: {random_value}")
 
    if random_value >= 5:
        print("랜덤 값이 5 이상입니다. task_A를 실행합니다.")
        return "task_A"
    else:
        print("랜덤 값이 5 미만입니다. task_B를 실행합니다.")
        return "task_B"
 
 
# 브랜치 태스크 (랜덤 값 생성 및 분기 결정)
branch_task = BranchPythonOperator(
    task_id="branch_task",
    python_callable=generate_random_and_branch,
    dag=dag,
)
 
# Task A (랜덤 값이 5 이상일 때 실행)
task_A = BashOperator(
    task_id="task_A",
    bash_command='echo "Task A 실행됨 - 랜덤 값이 5 이상입니다!"',
    dag=dag,
)
 
# Task B (랜덤 값이 5 미만일 때 실행)
task_B = BashOperator(
    task_id="task_B",
    bash_command='echo "Task B 실행됨 - 랜덤 값이 5 미만입니다!"',
    dag=dag,
)
 
# 태스크 의존성 설정
branch_task >> [task_A, task_B]

위 DAG를 실행해서 나온 랜덤 값이 1이라서 5보다 작기 때문에 task_B가 실행되었다

Trigger Rules

일부만 성공해도 다음 task를 실행하는 등 다양한 Trigger rules를 인자로 넣어서 자유롭게 커스터마이징이 가능하다

• all_success (기본값): 모든 upstream 성공 시 실행
• all_failed: 모든 upstream 실패 시 실행
• all_done: 모든 upstream 실행
• all_skipped: 모든 upstream 이 skipped 상태인 경우 실행
• one_success: 하나라도 성공하면 실행
• one_failed: 하나라도 실패하면 실행
• one_done: 하나라도 성공 혹은 실패하는 경우 실행
• none_failed: 실패만 없으면 실행 (성공/스킵 혼합 허용)
• always: 무조건 실행
• …

from airflow.operators.empty import EmptyOperator
 
notify = EmptyOperator(
	task_id="notify",
	trigger_rule="one_failed"
)

Latest Only

예를 들어 한 달치 데이터를 처리하기 위해 과거 날짜에 대한 backfill이 필요한 경우를 생각해보자
알림 전송 및 리포트 생성 작업의 경우 과거 데이터에는 굳이 진행하지 않아도 되는 작업일 것이다

이런 경우 Latest Only를 사용해 최신 DAG 실행에만 특정 태스크를 실행하도록 제어할 수 있다

LatestOnlyOperator를 통해 그 이후 downstream task들을 Latest DAG run 이 아닌 경우 자동으로 skip할 수 있다

Latest DAG run? → 현재 시간이 헤당 DAG 실행 시간과 다음 스케줄 사이에 있고, 외부에서 수동으로 trigger 된 실행이 아닌 경우

오늘 날짜가 2025-09-05이고 catchup=True를 이용해서 2025-09-01부터 오늘까지 작업을 모두 처리하게 DAG를 작성한 예시이다

import pendulum
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.latest_only import LatestOnlyOperator
from airflow.sdk import DAG
from airflow.utils.trigger_rule import TriggerRule
 
with DAG(
    dag_id="latest_only_dag",
    schedule="@daily",
    start_date=pendulum.datetime(2025, 9, 1, tz="UTC"),
    catchup=True,
    tags=["example3"],
) as dag:
    latest_only = LatestOnlyOperator(task_id="latest_only")
    task1 = EmptyOperator(task_id="task1")
    task2 = EmptyOperator(task_id="task2")
    task3 = EmptyOperator(task_id="task3")
    task4 = EmptyOperator(task_id="task4", trigger_rule=TriggerRule.ALL_DONE)
 
    latest_only >> task1 >> [task3, task4]
    task2 >> [task3, task4]

가장 최근인 오늘을 제외하고 이전 데이터 처리의 경우 LatestOnlyOperator로 인해 task1과 task3가 skip된 것을 확인할 수 있다

Depends On Past

Task가 이전 DAG Run에서 성공했을 때만 현재 실행을 허용하는 기능으로 같은 Task가 이전 실행에서 실패했다면, 현재 실행도 자동으로 건너뛰거나 실패 처리된다

데이터 적재, ETL 등에서 연속성이 중요한 경우 (예: 하루치 데이터가 누락되면 다음날 데이터 적재도 막고 싶을 때) 혹은 Task가 순차적으로 실행되어야 할 때 (예: 파일 처리, 배치 작업 등) 에 활용된다

Setup and Teardown

Setup과 Teardown은 Airflow에서 리소스의 생성과 정리를 자동화하여, 데이터 파이프라인 실행 시 리소스 관리가 누락되지 않도록 도와주는 기능이다

---

3. Operator

Operator는 특정 작업을 수행하도록 설계된 재사용 가능한 Task 템플릿 이다

지금까지 예시에서 사용된 EmptyOperator, BashOperator, PythonOperator 등과 같이 원하는 기능을 편하게 사용할 수 있게 도와주는 역할을 한다
많이 사용하는 대부분의 Operator는 Airflow에서 기본으로 제공하거나, 혹은 python package만 설치(참고 : Provider)하면 쉽게 사용할 수 있다

대표적인 Operators

• EmptyOperator : 아무 작업을 하지는 않지만, 테스트나 작업 간의 의존성을 명확히 할 때 사용할 수 있음
• BashOperator

  run_this = BashOperator(
      task_id="run_after_loop",
      bash_command="echo https://airflow.apache.org/",
  )

• PythonOperator : 하고 싶은 작업을 파이썬 함수로만 작성하면 끝

  def print_context(ds=None, **kwargs):
  """Print the Airflow context and ds variable from the context."""
  print("[group] All kwargs")
  pprint(kwargs)
  print("[endgroup]")
  print("[group] Context variable ds")
  print(ds)
  print("[endgroup]")
  return "Whatever you return gets printed in the logs"
   
  run_this = PythonOperator(task_id="print_the_context", python_callable=print_context)

• BranchDateTimeOperator : DAG 실행 시간에 따라 branch 작업을 구성하고 싶은 경우 사용

  empty_task_11 = EmptyOperator(task_id="date_in_range", dag=dag1)
  empty_task_21 = EmptyOperator(task_id="date_outside_range", dag=dag1)
   
  cond1 = BranchDateTimeOperator(
      task_id="datetime_branch",
      follow_task_ids_if_true=["date_in_range"],
      follow_task_ids_if_false=["date_outside_range"],
      target_upper=pendulum.datetime(2020, 10, 10, 15, 0, 0),
      target_lower=pendulum.datetime(2020, 10, 10, 14, 0, 0),
      dag=dag1,
  )
   
  # Run empty_task_11 if cond1 executes between 2020-10-10 14:00:00 and 2020-10-10 15:00:00
  cond1 >> [empty_task_11, empty_task_21]

• EmailOperator
• HttpOperator
• SQLExecuteQueryOperator
• DockerOperator
• HiveOperator
• S3FileTransformOperator
• SlackAPIOperator
• …

Custom Operator

공식적으로 제공되지 않지만 자주 사용하는 기능이 있는 경우 Custom Operator를 직접 만들어 사용할 수 있다 내장 Operator로 표현하기 어려운 로직을 BaseOperator 서브클래스로 캡슐화한 사용자 정의 템플릿이고 Airflow 3에서는 공식 SDK 경로 airflow.sdk 의 BaseOperator 상속을 권장하고 있다

원하는 Operator를 만들기 위해서는 BaseOperator 클래스를 상속 받아 두 가지 메서드를 override 해야한다

• 생성자 (__init__) : Operator에 필요한 파라미터 지정. 이 때 DB 호출과 같은 무거운 작업은 금지
• 실행 method (execute) : Operator가 실행될 때 호출되는 코드. Airflow context를 파라미터로 받아 설정값을 읽을 수 있음

왜 __init__에서 무거운 작업이 금지?

• Scheduler가 DAG 파일 주기적으로 parsing하며 각 Task에 해당하는 Operator Instance 계속 생성 → __init__ 호출
  • 따라서 생성자에서 DB연결, API 호출, 대용량 파일 읽기 등을 수행하면 비효율적
  • 가볍게 파라미터 저장만 하는게 효율적
• execute는 실제 실행될 때 한 번 호출되기 때문에 외부 리소스 접근은 execute에서 해야함
  • 따라서 Hook, Client 생성, 인증 조회, 쿼리/요청 전송 등은 execute에서 생성 및 호출

Custom Operator를 정상적으로 import해서 사용하기 위해서 Airflow의 PYTHONPATH에 포함된 폴더(dags/, config/, plugins/)에 두어야 한다
일반적으로 plugins/ 폴더에 많이 작성하고 아래 예시도 plugins/ 에 추가하면 된다

# plugins/hello_operator.py
from airflow.sdk import BaseOperator  # Airflow 3 공개 SDK
 
class HelloOperator(BaseOperator):
	def __init__(self, name: str, **kwargs):
		super().__init__(**kwargs)  # task_id 등 공통 인자 처리
		self.name = name
 
	def execute(self, context):
		message = f"Hello, {self.name}!"
		print(message)
		return message

# DAG에서 사용
from datetime import datetime
from airflow import DAG
from hello_operator import HelloOperator
 
with DAG(dag_id="custom_op_example", start_date=datetime(2023, 1, 1), schedule=None, catchup=False) as dag:
	hello = HelloOperator(task_id="hello_task", name="Airflow 3")

---

4. Sensor와 Deferrable Operator

Sensor

Airflow에서 Sensor는 특정 조건이 만족될 때까지 대기(polling or trigger) 하는 역할을 담당하는 특수한 Operator로, 조건이 충족되면 다음 Task 실행이 이어지도록 한다

대표적인 Sensor는 다음과 같다

• BashSensor: 지정한 Bash 명령어 또는 스크립트가 성공 (리턴 코드 0)할 때까지 대기
• TimeDeltaSensor: 설정한 시간 (delta)이 경과할 때까지 대기
• TimeSensor: 지정한 시각 (target_time)이 될 때까지 대기
• DayOfWeekSensor: 지정한 요일이 될 때까지 대기
• FileSensor: 특정 경로에 파일이 존재할 때까지 대기
• PythonSensor: 지정한 파이썬 함수가 True를 반환할 때까지 주기적으로 실행하며 대기
• ExternalTaskSensor:  다른 DAG (또는 같은 DAG의 다른 실행 시점)의 특정 작업이 완료될 때까지 현재 작업을 대기

Sensor에는 대기 방식에 따라 두 가지 모드가 존재한다

• poke (기본): 조건을 주기적으로 확인하면서 Worker 슬롯을 계속 점유
• reschedule: 조건 확인 후 일정 시간 대기하고 Worker 슬롯을 반납, 다음에 재스케줄

보통 짧은 대기에는 poke, 긴 대기에는 리소스 효율적이므로 reschedule 을 사용하는 것이 바람직하다

다음은 특정 경로에 ‘sensor_test.txt’라는 파일이 생성되는 경우 success가 되는 예시이다

import pendulum
from airflow import DAG
from airflow.providers.standard.sensors.filesystem import FileSensor
 
with DAG(
    dag_id="ex_file_sensor",
    start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
    schedule="@daily",
    catchup=False,
) as dag:
    wait_file = FileSensor(
        task_id="wait_for_file",
        filepath="/opt/airflow/plugins/sensor_test.txt",
        mode="poke",
        poke_interval=30,  # 30초 간격으로 체크
        timeout=60 * 60 * 2,  # 2시간 대기 타임아웃
    )

이 DAG를 실행하기 전에 local filesystem 접근을 위한 connection을 만들어야 한다

• connection 이름 : ‘fs_default’
• connection type : File (path)
• Path : ’/’

fs_default라는 이름은 FileSensor가 기본값으로 인식하는 conn_id이고 Path를 ’/‘로 해서 filepath에는 절대경로를 입력해주면 된다

이 DAG를 실행하면 마운트된 /opt/airflow/plugins 폴더에 sensor_test.txt라는 이름의 파일이 생길 때까지 계속 running 상태로 남아있고 해당 파일이 추가되면 success 상태가 된다

Deferrable Operator

Deferrable Operator는 Airflow의 Operator 또는 Sensor가 외부 조건을 기다릴 때 리소스 효율을 극대화하기 위해 고안된 특별한 유형으로 동작 원리는 다음과 같다

• Deferrable Operator가 특정 조건을 기다려야 할 때, trigger 객체를 통해 실행을 중단
• Trigger는 Airflow의 triggerer 컴포넌트에서 비동기로 동작하며, 조건이 충족될 때까지 감시
• 조건이 충족되면 트리거가 신호를 보내 Operator가 다시 실행을 재개

기존 Operator와 Sensor (poke 모드)는 작업이 대기 상태여도 worker slot을 계속 점유하지만, Deferrable Operator는 대기 중에 worker slot을 해제하고 trigger가 대기 로직을 대신 처리하게 된다

Deferrable Operator의 장점은 다음과 같다

• 리소스 절약: 긴 대기 시간이 필요한 작업에서도 worker 자원을 점유하지 않음  
• 확장성 개선: 수백~수천 개의 Sensor가 동시에 대기하더라도 Triggerer 프로세스가 관리하므로 클러스터 효율이 높음  
• 운영 안정성: 장기 대기로 인해 worker가 묶이지 않으므로 SLA 지연이나 큐 적체를 방지

보통 기존 sensor, operator 이름 뒤에 Async가 붙어있거나 혹은 deferrable=True라는 파라미터를 설정해주면 deferrable 모드로 동작하게 된다

Deferrable Operator 사용 시 몇 가지 주의해야할 점이 있다

• Triggerer 프로세스 실행 필요: airflow triggerer 프로세스가 반드시 구동 중이어야 함  
• 리소스 한계 고려: Triggerer가 모든 deferrable 태스크를 감시하므로, 너무 많은 deferrable 태스크가 한꺼번에 몰리면 triggerer 튜닝 필요  
• 호환성: 모든 Operator가 deferrable 버전을 제공하는 것은 아니며, 일부는 별도의 provider 패키지에 포함됨

마지막으로 Sensor에서 reschedule 모드와의 차이를 정리하면 다음과 같다

구분	reschedule 모드	Deferrable Operator
Worker 점유	조건 확인 후 worker 슬롯을 반납하지만, 다시 잡혀서 실행됨	worker 슬롯 완전 해제, Triggerer가 감시
대기 처리 주체	Scheduler가 주기적으로 태스크를 rescheduling	Triggerer 프로세스가 이벤트/조건을 비동기 감시
리소스 효율성	worker 점유는 줄지만 주기적 rescheduling으로 metadata DB·스케줄러 부하 존재	장기 대기에 최적화, metadata DB 접근 최소화, 대규모 sensor에도 안정적
지원 여부	대부분의 Sensor에서 mode="reschedule" 옵션만 설정하면 사용 가능	전용 Async/Deferrable Operator 필요
적합한 상황	수분~수십 분 대기, Sensor 수가 많지 않을 때	수시간~수일 이상 장기 대기, 수백 개 이상의 Sensor 병렬 실행 시

---

5. Hook

Airflow에서 Hook은 외부 시스템(DB, API, 클라우드 서비스 등)과 연결하기 위한 고수준 인터페이스다.   BaseHook 클래스를 상속하여 구현하며 Airflow Connections 정보를 사용해 인증, 엔드포인트, 포트 등을 자동으로 가져올 수 있다

Hook이 필요한 이유는 다음과 같다

1. 일관된 인터페이스 제공
   • 다양한 DB/API를 동일한 방식으로 접근 → 코드 가독성, 유지보수성 ↑
2. Connection 관리 통합
   • Airflow UI에서 Connection을 설정하면 Hook이 이를 자동으로 불러옴  
   • 비밀번호, 토큰 같은 민감 정보는 코드에 노출되지 않음
3. 모듈화·재사용성  
   • 여러 DAG에서 동일 Hook을 불러와 활용 가능  
   • 커스텀 Hook 작성 시, 팀 차원의 공통 유틸처럼 사용 가능
4. 복잡한 작업 처리  
   • Operator 수준에서 제공하지 않는 기능을 Hook으로 직접 구현 가능  
   • 하나의 Task에서 두 개 이상의 Hook을 병행해 사용할 수도 있음

Operator와 비교하면 다음과 같다

• Operator
  • Airflow에서 실행 단위(Task)를 정의  
  • 내부적으로 Hook을 호출하여 외부 시스템과 상호작용  
• Hook  
  • 외부 연결을 추상화  
  • Operator에서 사용되는 하위 구성 요소  
  • 직접 실행되지 않음 (Task가 아님)

Operator는 “무엇을 실행할지”, Hook은 “어디에 연결할지”를 담당한다고 볼 수 있다.

Connection

보통 Hook다른 서비스에 연결을 할 때 사용하기 때문에 연결 정보 가 필요한데 이를 Airflow Connection에 등록해서 사용할 수 있다

아래의 예시처럼 Airflow UI에서 직접 등록하거나 아예 환경변수로 docker compose에 주입할 수 있다

예시: PostgresHook

Airflow UI에서 Admin>Connection>Add Connection에 들어가서 아래와 같이 ‘postgres_default’라는 connection 이름으로 Airflow metaDB로 사용 중인 postgres와 연결할 수 있다

그 후 아래 DAG를 실행하면 ‘postgres_default’ connection을 이용해 PostgresHook이 Postgres에 접속하게 되고 “SELECT now()::timestamp, current_database()” 라는 쿼리를 실행할 수 있다

import pendulum
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
  
def fetch_rows(**_):
    hook = PostgresHook(postgres_conn_id="postgres_default")
    rows = hook.get_records("SELECT now()::timestamp, current_database()")
    print(rows)
 
with DAG(
    dag_id="ex_hook_postgres",
    start_date=pendulum.datetime(2025, 1, 1, tz="Asia/Seoul"),
    schedule="@daily",
    catchup=False,
    tags=["hook", "postgres"],
) as dag:
    read_pg = PythonOperator(
        task_id="read_postgres",
        python_callable=fetch_rows,
    )

Hook도 Operator와 마찬가지로 BaseHook을 상속받아 Custom으로 직접 작성할 수 있다

---

6. Provider와 Plugin

Provider

Provider는 특정 서비스나 플랫폼과의 통합을 패키지 단위로 제공하는 확장 모듈 집합으로 Operator, Hook, Sensor, Transfer, 연결 타입, 로그 핸들러, 시크릿 백엔드, 알림 등 다양한 확장을 포함할 수 있다

아주 다양한 Provider 패키지가 있기 때문에 간단하게 airflow 환경에 install해서 자유롭게 추가기능을 활용할 수 있다

예를 들어 airflow를 통해 spark 관련 작업을 하고 싶을 때 apache-airflow-providers-apache-spark 를 사용하면 된다

설치 방법은 다음과 같다

• airflow dockerfile에 해당 python패키지를 추가해서 다시 빌드해서 사용 (권장)
• .env에 _PIP_ADDITIONAL_REQUIREMENTS='apache-airflow-providders-apache-spark'를 추가하고 docker compose 실행 (임시 개발용)

해당 provider docs를 보면 Connection types, Decorators, Operators 등 이 패키지에서 제공하는 기능들을 모아서 볼 수 있다

다음 docs를 참고해 custom provider를 직접 만들 수 있다

Plugin

Plugin은 Airflow의 동작과 UI를 확장/변경하기 위한 메커니즘이다

앞서 custom operator를 만들 때 봤듯이 plugins/ 직접 기능을 추가할 수 있는 것이라 볼 수 있는데
단순히 operator, hook 등을 추가하는 것을 넘어서 UI, Flask 뷰, Admin 화면 등 Airflow의 자체 기능에 대한 custom도 가능하다

즉, Provider가 특정 서비스와의 연동을 위한 공식 배포 모듈이라면 Plugin은 UI, Flask 확장까지 가능한 사용자 정의 모듈로 Provider를 포괄하는 개념이다

당연히 직접 custom plugin을 구현할 수 있으며, airflow-executor처럼 패키지화 된 plugin도 있다
DAG/태스크 상태를 Prometheus로 노출하는 플러그인으로, 해당 패키지를 설치하면 http://localhost:8080/admin/metrics 경로가 추가되어 Prometheus에서 이를 인식해서 모니터링을 편하게 할 수 있다

---

7. 주요 개념 비교

개념	목적/역할	확장성	사용 방법
Task	DAG 내 실행 단위로서 Operator 인스턴스 또는 @task 함수가 생성하는 노드	자체 확장 개념은 없음, 템플릿(Operator) 다양화로 형태 확장	DAG에서 Operator 인스턴스 생성 또는 @task 사용, 의존성 연결
Operator	재사용 가능한 작업 템플릿으로 execute()에 수행 로직 정의	BaseOperator 상속으로 커스텀 구현 가능	DAG에서 클래스를 인스턴스화하여 Task 생성
Sensor	외부 조건(파일 존재, API 응답, 특정 시각 등)을 만족할 때까지 대기하는 특수 Operator	BaseSensorOperator 상속, Deferrable 버전 구현 가능	mode="poke" 또는 mode="reschedule"로 사용, 일부는 Async/Deferrable Sensor로 사용
Hook	외부 시스템 인터페이스, Connection과 연계된 고수준 API 제공	BaseHook 상속으로 커스텀 Hook 구현 가능	보통 Operator 내부에서 생성·호출, 필요 시 PythonOperator에서 직접 사용
Provider	특정 서비스 연동을 패키지 단위로 제공, 코어 외 확장 묶음	코어와 독립 버전, 신규/사내 Provider 제작·배포 가능	pip install apache-airflow-providers-... 설치 후 모듈에서 클래스 임포트
Plugin	Airflow 동작/UI 확장을 위한 플러그인 시스템	Flask/FastAPI 뷰, 리스너, 매크로 등 전역 확장 구성 가능	plugins/ 폴더 또는 패키지 엔트리포인트로 등록 후 재시작