[Apache Airflow] Airflow Architecture

---

Airflow Architecture

• Architecture
• Executors

---

Architecture

• Scehduler: Triggers schedule workflows and submits tasks for executor to run
• Executor: Defines how to run tasks depending on the system
• Worker: Runs the actual tasks
• Metadata database: Stores information about DAGs and tasks states
• Webserver: User interface to insepct, trigger and debug DAGs and tasks behavior
• Dags folder: Directory where all DAGs code is persisted, read by scheduler & executor

Architecture

Executor-Task

---

Executors

LocalExecutor

• LocalExecutor의 경우 단일 노드의 Scheudler에 의존하기에 Production에 적합하지 않음

CeleryExecutor

• CeleryExecutor의 경우, Worker가 항상 일하기에 운영하는 Dag가 많을 경우 더 적합할 수 있음
• Celery Queue는 Result Backend와 Broker로 이루어짐
  • Result Backend: worker가 exectue한 task의 status를 저장
  • Broker: scheduler가 수행해야될 task를 보내는 queue이자, worker가 수행해야될 task를 pull해오는 queue
    • 가령 한 Dag가 T1 >> [T2, T3] >> T4 로 되어있는 경우가 존재
      • 1) Scheduler는 T1을 Broker에게 보냄
      • 2) Worker1이 Broker에서 T1을 받아온 뒤 수행 완료 후 Result Backend에 결과 완료여부를 보냄
      • 3) Scheduler는 T2, T3를 Broker에게 보냄
      • 4) Worker2이 T2를, Worker3가 T3를 받아온 뒤 각각 수행을 완료한 후 Result Backend에 결과 완료여부를 보냄
      • 5) Scheduler는 T4를 Broker에게 보냄
      • 6) Worker1이 Broker에서 T4을 받아온 뒤 수행 완료 후 Result Backend에 결과 완료여부를 보냄
    • 특정 Worker에서 특정 Task가 보내지도록 Worker마다 Queue를 설정할 수 있음
      • 이 경우 여러 개의 Queue가 존재하게 됨
      • Worker마다 사양(CPU/ GPU)이 다를 수 있기 떄문에 유용
• cfg 설정 사항
  • 1) executor=CeleryExecutor
  • 2) sql_alchemy_conn=postgresql+psycopg2://[user]:[pw]]@[host]/[db]
  • 3) celery_result_backend=postgresql+psycopg2://[user]:[pw]]@[host]/[db]
    • 반드시 airflow의 metadata를 store하는 db랑 같은 db를 써야되는 것은 아님
  • 4) celery_broker_url=redis://:@redis:6379/0
    • 위는 redis를 사용하는 경우의 예시, RabbitMQ도 가능
• Worker가 될 컴퓨터에서 celery worker라고 입력하면 됨
  • 만약 해당 Worker에만 보내고 싶은 Task가 있는 경우 celery worker -q [queue_name] 입력
    • (e.g. celery worker -q high_cpu)
    • 🔗 특정 task를 특정 queue를 보내고 싶은 경우
    • cf. 기본적으로 모든 task는 default queue로 감

KubernetesExecutor

• 위와 같은 경우가 아니라면 Kubernetes를 활용할 떄 더 효율적으로(cost-effective) pipeline 운영 가능

---

ref

• 🔗 참고