Airflow

https://developers.hyundaimotorgroup.com/blog/570

(Airflow #1) 데이터 엔지니어들이 선택하는 Apache Airflow 소개

위 블로그를 참고해서 작성하였다.

 

Apache Airflow 는 Airbnb 에서 workflow 들을 관리하고 스케줄링 하기 위해 만든 파이썬 기반의 오픈 소스로

Workflow 를 Python code 로 작성할 수 있으며, DAG(Directed Acyclic Graph) 라는 대분류 안에 workflow 들이 속하여 스케줄링하고, DAG를 시각화해서 보여준다

 

왜 Airflow 를 사용하는지?

오픈소스이다, python기반으로 작성한다, 쿠버네티스 지원한다(hadoop yarn지원안함)

 

helm차트 활용 가능

Helm 차트는 Kubernetes에서 애플리케이션을 템플릿 기반으로 배포 자동화할 수 있게 해준다.

따라서 Airflow는 복잡한 컴포넌트(webserver, scheduler, worker 등)가 있어서 수동 설치하지 않고, Helm 차트 덕분에 버전 관리, 구성 설정, 재배포를 한 번에 관리할 수 있어 DevOps 작업 부담이 크게 줄어든다.

 

배치 플랫폼 구축에 용이

Airflow는 시간 기반, 의존성 기반, 이벤트 기반 배치 모두 가능.

병렬 실행, 의존성 관리, 재시도, 알림, 모니터링을 기본 제공.

기존 스크립트나 Spark/Shell/Python 작업을 재활용 가능.

즉, 코드 몇 줄로 복잡한 스케줄러를 쉽게 구축 가능하여 타 배치 시스템 대비 개발/운영 효율적이다.

 

XCOM을 통해 task 간 정보 전달 용이

일반적인 배치 작업은 task 간 의존성뿐만 아니라 데이터 전달도 필요.

Airflow의 XCom을 쓰면 값을 메타데이터 DB에 자동 저장하고 다음 task에서 바로 꺼내 씀.

따로 파일 저장하거나 Redis 같은 메시징 시스템 쓸 필요 없이 내장 시스템으로 공유 가능.

 

pool을 지정하여 스케줄 관리 가능

작업이 많을 때 리소스가 몰리면 서버가 터진다.

pool을 지정하면 작업 그룹 별로 동시 실행 개수 제한 가능하여 자원 과부하 방지와 우선순위 제어를 통해 안정적 운영 가능

 

sensor나 trigger를 활용해 이벤트에 의한 배치 flow가능.

단순히 "매일 몇 시에 실행"하는 걸 넘어서, S3에 파일이 올라오면 처리, Kafka 메시지가 오면 처리 이런 이벤트 기반 배치 작업이 가능하다.

즉, 불필요한 polling 줄이고 즉각적 반응형 워크플로우 구성 가능하고, 특히 실시간 데이터 흐름이 있는 환경(금융, IoT, 물류 등)에서 매우 유용

 

airflow는 3가지 형태로 구성

Basic Airflow deployment

기본적으로 Airflow 를 설치할 때 필수로 구성되는 아키텍처 입니다.

Distributed Airflow architecture

흔히들 설치 시 많이 따르고 있는 아키텍처 입니다.

 

Separate DAG processing architecture

DAG Processor 를 보안상의 이유로 별도로 분리한 아키텍처 입니다. 기존의 scheduler 에서 DAG 파일을 읽어 들였다면, 해당 아키텍처에서는 분리하여 관리되게 구성할 수 있습니다.

 

 

필수 Component 구성

scheduler : Workflow 작업을 모두 처리하는 역할을 합니다. 기본적으로 여러 가지 실행 프로그램을 사용할 수 있으며 직접 작성할 수도 있습니다.

webserver : 사용자가 trigger, 디버깅, DAG 실행 상태 등을 확인할 수 있는 웹 인터페이스 입니다.

DAG 파일 폴더 : 스케줄러가 실행 시킬 workflow 들을 읽어 들이는 경로 입니다.

metadata database : workflow 와 task 의 상태를 저장하는 airflow component 입니다. 메타데이터 데이터베이스 설정은 Set up a Database Backend에 설명되어 있으며 Airflow가 작동하는 데 필요합니다.

 

선택적 Component 구성

worker : 스케줄러에 의해 task 들을 실행 시키는 역할을 합니다. 기본 구성에서는 scheduler 와 구분되지 않고 스케줄러가 worker 동작을 합니다. Worker는 CeleryExecutor 의 process 나 KubernetesExecutor 의 POD 로 장기적 실행할 수 있습니다.

triggerer : asyncio 이벤트 루프에서 지연된 작업을 실행합니다. 지연된 작업을 사용하지 않는 기본 설치에서는 트리거가 필요하지 않습니다. 작업 연기에 대한 자세한 내용은 Deferrable Operators & Triggers 에서 확인 가능합니다.
dag processor: DAG 파일을 구문 분석하고 메타데이터 데이터베이스로 직렬화 합니다. 기본적으로 DAG processor 프로세스는 스케줄러의 일부이지만 확장성과 보안상의 이유로 별도의 구성 요소로 실행될 수 있습니다. DAG processor 있는 경우 스케줄러는 DAG 파일을 직접 읽을 필요가 없습니다. DAG 파일 처리에 대한 자세한 내용은 DAG File Processing 에서 확인할 수 있습니다.
folder of plugins : 플러그인은 Airflow의 기능을 확장하는 방법입니다(설치된 패키지와 유사). 플러그인은 Scheduler, DAG processor, Trigger 및 Webserver 에서 읽습니다. 플러그인에 대한 자세한 내용은 Plugins 에서 확인할 수 있습니다.

Airflow Core concepts

Airflow 는 DAG 들을 관리하며, DAG 마다 각각 Task 들의 흐름들로 이루어져 있습니다. Task는 각각 Operator 로 구성됩니다. 

DAG
DAG(Directed Acyclic Graph | 방향성 비순환 그래프)는 작업의 흐름이 다시 동작했던 작업으로 돌아오지 않고 방향성을 가지고 끝을 향해 처리하는 형태를 말합니다. Airflow 에서 DAG 는 Task 들이 모여 여러 작업 흐름의 집합으로 보면 좋을 듯 합니다.

Task
Task 는 Airflow 의 기본 실행 단위 입니다. DAG 안에서 순서를 가지고 실행 됩니다. 아래와 같이 Python 코드 내에서는 변수로 선언되며, Operator 로 선언 됩니다.

operator
Task 를 정의할 때 사용하는 템플릿입니다.  많은 Operator 를 제공하고 있으며 BaseOperator 를 상속받아 Custom Operator 를 만들어 사용할 수도 있습니다.

Executor
Task 가 실행되는 인스턴스 입니다. Operator 로 Task 를 선언하는 부분에서 지정할 수도 있지만 보통은 Config 파일에 지정된 Executor 로 실행시킵니다. Executor 종류는 아래와 같이 있습니다.

Local Executors (로컬):

• Local Executor
• Sequential Executor

Remote Executors (외부):

• CeleryExecutor
• BatchExecutor (AWS)
• KubernetesExecutor
• EcsExecutor (AWS)

XComs
XComs(cross-communications) 는 Task 간 소통할 수 있게 하는 방법 입니다. 각 Task 별로 XCom 키-값 형태로 다음 Task 가 사용할 정보를 저장할 수 있습니다.