Airflow Summit 2021 - the Newcomer's guide to airflow's architecture

https://youtu.be/oLTMN-4Rvj8?si=ShQgVv-1M3ZlPbDG

 

 

Airflow는 AirBnb 내부에서 사용할 목적으로 만든 internal ETL tool에서 시작함.

• 처음부터 훌륭한 아키텍처로 구성된 건 아니었고, 사용자가 많아지면서 요구사항에 대응하는 식으로 개선되어간 Organic Product.

 

Airflow에서는 사용자가 DAG을 작성한다.

• DAG: 해야 할 Task와, task 간 relationship을 정의하는 Template.
  • operators / task를 python으로 작성
• DAG을 실제로 실행할 때는 용어가 조금 달라진다.
  • DagRun: instance of DAG that runs a certain execution time. 즉 특정 파라미터와 데이터를 받아서, 일정 기간 실행되는 DAG 인스턴스
  • 공식문서에 operator / task가 혼용되어 쓰이는데
    • Operator: "template for a task".
    • operator에 parameter를 전달하면 instantiating it into a "Task".
  • TaskInstance: DAG을 실행해서 DAGRun으로 바뀌고 operator(template)에 parameter가 전달되어 Task가 만들어지면, 그걸 TaskInstance 라고 한다

이 용어정리가 꽤 중요한 이유는, 예컨대 TaskFlow API를 호출하게 되면 Operator -> Task 변환 작업을 건너뛰는 것이기 때문. 이 경우 Task를 직접 정의해줘야 한다.

• Operator는 "Easy Task Template"

 

Airflow는 근본적으로 2 Projects in One 구조.

• Scheduler: DAG 정보와 현재 schedule 상태, dependencies between tasks를 확인해서, 아래 세 가지를 결정함
  • What To Run
  • When it should Run
  • What Conditions it should run on
• scheduler의 output: List of things that needs to run now. (task Instances)
• output 결과를 Executor가 받아서, 어느 executor가 taskinstance를 실행할지 결정한다.
  • Orchestrates how to run things.

 

논리적으로는 Scheduler -> Executor 다이어그램이 단순하지만, 실제 컴포넌트 구성방식은 위의 구조에 가깝다.

• scheduler 내부에 executor가 있다.
  • localExecutor의 경우, scheduler에서 할당한 DAG file을 로컬에서 실행한다.
• Web Server: UI 관련 작업
• Database: metadata 정보, 어떤 task instance와 DagRun이 어디서 동작하는지를 기록.
• DAG files: 사용자가 작성한 templates
  • 따라서, State를 저장하는 시스템은 DB와 File system 두 개.

 

Celery Executor 방식

• scheduler는 실행해야 할 task를 Celery Executor에 전달.
  • celery는 Task Queue의 Abstraction (redis / rabbitmq). 실행할 task를 Queue에 전달한다
  • Workers는 task queue의 메시지를 pull 받고, 필요한 DAG file을 실행한다.

이 경우 Main Component는 Scheduler (With Celery Executor), Web Server, Worker 세 개.

 

Executor는 일종의 interface. 기본적으로는 scheduler에 내장된 채 실행되지만, interface만 만족하면 외부에서도 실행 가능하다.

Airflow는 Metadata 저장된 DB에 매우 의존적이다. postgres 등 성능과 안정성이 검증된 걸 사용하는 게 좋음.

• 컴포넌트 간 통신하는 프로토콜이 따로 없고, 전부 DB를 바라보고 있다.
• SPOF 요소이기도 하지만, Single Point of Coordination이기도 함.
  • 모든 컴포넌트가 서로 통신하는 것보다는 단순한 아키텍처
• Locking, Transaction 사용해서 DB에 접근하는 모든 컴포넌트가 consistent View를 보장하도록 함.

Airflow 2.0 이후부터는, main component 세 개; Scheduler, Worker, WebServer 셋 다 HA 구조.

• DB의 경우 Active-Passive Setup 등의 방식으로 Availability를 확보해야 한다.

Scheduler

https://youtu.be/DYC4-xElccE?si=VUq-46aRQwlHgUMw 소개를 참고하라고 하고 넘어감.

Executor

interface만 충족하면 pluggable하게 쓸 수 있음. Celery / Kubernetes, 또는 둘 다 사용할 수도 있음.

CeleryExecutor

• task는 Queue로 전달되고, worker가 queue에서 pull받아 실행한다.
• worker는 consistent process이며, 한 번에 하나의 작업만 실행한다.
  • do not restart between tasks
• isolation이나 보안 면에서는 허점이 있지만, bootup 시간이 없기 때문에 전반적인 처리속도가 빠름.

 

k8s Executor의 경우, 하나의 task instance가 pod에 대응됨.

• isolation 면에서는 강점, 매번 pod를 생성하는 구조이므로 bootup 시간이 필요.

 

How Executors do handle task instances? 를 소개하는 다이어그램. 복잡해 보이지만, Blue Box 기준으로 이해하면 된다.

• Scheduler: DAG를 실행할 때 (= DAGRun 생성)
  • DAGRun 내부 모든 task instances의 초기 상태는 없음. (No Status == Unscheduled)
  • previous Dependencies가 전부 해결되었을 경우, scheduled 로 변경
• Executor: task가 scheduled된 이후의 동작을 관장함.
  • task를 sucesss / fail로 변경하는 역할.
  • Queue와 worker 구조일 경우, worker가 역할 수행.
  • fail시 retry 수행, suspend와 같은 기능도 포함됨.

 

airflow는 Task (task instance) 단위로 이루어진다.

• DAG는 grouping mechanism에 가까움. 내부 코드를 봐도, input 단위로 쓰이긴 하지만 core part of scheduling과는 거리가 있다.
• 따라서 task가 어떤 flow를 따르는지 / dependency before & after task 확인하는 게 좋다

 

flexible environment: 무엇이든 실행할 수 있다. 장점이자 단점이지만, 장점으로서의 기능이 더 크다고 생각

What can we improve?

 

• Async, Eventing 기능. trigger-based invoking 기능
• 모든 로직이 DB Connection을 연결하는 구조. API가 확장성 면에서 훨씬 낫기 때문
• 아키텍처 / 문서 보강 필요
• 최고의 아키텍처를 구성하는 게 목적은 아님. Maintenance, Backward Compatibility 지원을 중요시하고 있다.