[Airflow] Airflow Flower | Hook | Task Group | Xcom | Branching | provider

 

Celery Executor사용시 Celery worker와 Tasks들의 상태를 모니터링할 수 있는 웹페이지이다.

 

Celery worker 의 Status를 볼 수 있고, 해당 워커를 클릭하면 pool, Broker, Queues, Tasks Limits, Config, System, Other등의 정보들을 확인할 수 있다.

 

Broker는 Redis의 정보가 보인다.

 

---

 

 

Hook은 특정 서버나 DB 등 에어플로우에서 외부시스템에 접속할 때 사용된다.

사용되는 Connection정보는 Connections에 저장해둔 정보를 사용한다.

 

---

 

Task Group

Task들을 그룹으로 묶을 수 있다.

 

비슷한 성격의 태스크들을 묶어서 마치 모듈화할 수 있고, 이를 다른 대그에서도 복붙해서 사용할 수 있다.

 

대그코드에서는 Task들을 task_group 으로 묶은 뒤 그룹 안에서도 task들의 실행순서를 정의하고,

대그 코드에서도 실행순서를 정의할 때 task_group명을 사용한다.

 

---

 

Xcom 

 

val = 42

context['ti'].xcom_push(key='my_key', value=val)

or

return val # --> xcom_push(key='return_value', value=val) 

 

 

val = context['ti'].xcom_pull(task_ids='t1', key='my_key')

 

 

혹은 아래와 같이 만들 수 있다.

def t1() -> Dict[str, any]:

  return {

    "my_val": my_val,

    "my_sentence": my_sentence

  }

 

def t2(data: Dict[str, Any])

 

 

 

---

Branching

 

A -> B -> C

          -> D

에서 B를 branch task로 만들어서 조건에 따라 분기할 수 있다.

from operator import eq
from airflow.sdk import dag, task

@dag()
def branch():

    @task
    def a():
        return 1

    @task.branch
    def b(val: int):
        if val == 1:
            return "equal_1()"
        return "different_than_1()"

    @task
    def equal_1(val: int):
        print(f"Value is equal to {val}")

    @task
    def different_than_1(val: int):
        print(f"Value is different than {val}")

    val = a()
    b(val) >> [equal_1(val), different_than_1(val)]

branch()

 

 

 

---

 

provider

 

 

Airflow에서 "provider"는 특정 서비스, 시스템, 혹은 기술과 Airflow를 통합하기 위해 제공되는 패키지나 모듈을 의미합니다.

좀 더 구체적으로 말하자면, Airflow provider는 다음과 같은 역할을 합니다:

1. Operator, Sensor, Hook 같은 컴포넌트 제공
   • 예를 들어, AWS, Google Cloud, Microsoft Azure, Slack, MySQL, PostgreSQL 등 다양한 외부 서비스와 통신하기 위한 Operator(작업 수행), Sensor(조건 감지), Hook(연결 관리) 등이 들어 있습니다.
2. Airflow와 외부 시스템 간 인터페이스 역할
   • Airflow 작업이 외부 API 호출하거나 데이터베이스에 연결할 때 provider 내 컴포넌트를 사용합니다.
3. 모듈화 및 유지보수 용이성
   • Airflow 코어에서 직접 모든 외부 연동 기능을 포함하지 않고 provider 패키지로 분리해 관리하기 때문에, 필요한 외부 서비스만 추가로 설치해서 사용할 수 있습니다.

예를 들어, apache-airflow-providers-google 패키지는 Google Cloud Platform 서비스와 연동할 때 필요한 Operator, Hook 등을 제공합니다.

따라서, Airflow에서 provider는 외부 시스템과의 연동을 위한 플러그인 같은 역할을 하는 구성 요소라고 이해할 수 있습니다.