🤷♀️
데이터 엔지니어링 직무를 알아보면서 가장 많이 접했던 용어는 '데이터 파이프라인'이었다. 그리고 이 데이터 파이프라인을 설계 하는데 가장 큰 파이를 차지하고 있는 Airflow를 심도있게 학습해야겠다고 생각했다.
Apache Airflow 기반의 데이터 파이프라인 (원제 : Data pipelines with Apache Airflow) 책을 기본으로 학습한 내용을 기록하고자 한다.
Airflow
Airflow는 데이터 파이프라인을 스케줄링하고 모니터링하기 위한 도구이다. Airbnb에서 사내 워크플로를 관리하려고 개발되었고 2016년 Apache 재단에 가입하여 현재 Apache 재단의 최고 레벨 오픈소스 프로젝트이다.
Apache Airflow github
https://github.com/apache/airflow
주요 기능
DAG을 사용한 그래프 기반 파이프라인 표현
Airflow는 데이터 파이프라인을 DAG로 시각화하여 제공한다.
절차적으로 표현된 스크립트 형식이 아닌 그래프로 표현하기 때문에 Task를 병렬로 처리할 수 있어 컴퓨팅 리소스를 효율적으로 활용할 수 있다.
DAG에 대한 설명은 아래에서 더 자세하게 다뤄본다.
또한 파이프라인을 작은 점진적인 태스크로 분리(=Task를 단계별로 세분화)할 수 있어 파이프라인의 중간 Task가 실패하더라도 전체를 다시 실행하는 것이 아니라 실패한 Task만 재실행하는 Backfill이 가능하므로 변화에 유연하고 효율적이다.
Backfill
데이터 파이프라인에서 이전에 실행되지 않은 작업들을 일괄적으로 실행하여 유실되었던 데이터를 가져온다. 이로 인해 데이터 파이프라인을 일관성 있게 유지할 수 있다.
Python 기반 Task작성
위에서 말했던 DAG는 DAG.py 형태로 Python으로 작성된다. DAG를 구성하는 Task들은 Python함수로 구현되며 함수 내에서 데이터 처리, 데이터 추출 및 변환 등의 원하는 작업을 수행할 수 있다.
코드를 통해 Task를 정의하기 때문에 다른 정적 방법으로 워크플로를 관리하는 데이터 파이프라인 도구들(Oozie, Luigi 등)보다 훨씬 더 동적인 Task 작성이 가능하고 유연한 파이프라인을 생성할 수 있다는 장점이 있다.
또한 Airflow는 계속해서 외부 데이터베이스나 클라우드 서비스를 연동할 수 있도록 확장기능을 개발하고 있어 여러 시스템 간에 데이터 프로세스를 결합할 수 있는 복잡한 데이터 파이프라인을 구축을 가능하게 한다.
모니터링 및 알림
Airflow는 모니터링이 가능한 웹 인터페이스를 제공한다.
DAG (Directed Acyclic Graph)
Airflow는 DAG을 사용하여 그래프 기반으로 파이프라인을 표현하여 높은 유연성과 효율성을 가진다.
DAG는 Task를 나타내는 노드(파란색 원)와 의존성을 나타내는 엣지(화살표)로 구성된다. 따라서 이 그래프를 예로 들어 설명하자면, 1번 Task에서 시작된 엣지가 2번과 3번 Task를 가리키고 있으로 2번과 3번은 1번이 완료되기 전까지는 실행되지 않는 1번 Task에 대한 의존성을 가진다.
또한 DAG는 순환과 반복을 허용하지 않는다.
왜 순환을 허용하지 않는가?
엣지는 의존성을 나타낸다고 했으므로 2번 Task는 1번 Task가 완료된 후에만 실행되는데, 만약 2번 -> 1번으로 가는 엣지가 추가되어 순환 실행이 되어버리면 1번과 2번이 상호의존성을 가지게 되어 명확한 실행 경로를 확인할 수 없게 되고, 태스크가 완료되지 못해 그래프가 동작하지 않는다. (닭이 먼저냐 달걀이 먼저냐 이런건가..)
Airflow의 구성요소
데이터 파이프라인 구현을 위해 DAG 파일을 작성하게 되면 DAG파일은 DAG를 모아놓은 폴더인 DAG Directory에 추가되고 주기적으로 Airflow의 구성 요소들에 의해 감지되고 실행된다.
Scheduler
DAG 파일을 분석하고 현재 시점에서 DAG의 스케줄이 지난 경우 Worker에 DAG의 Task를 예약한다.
Worker
예약된 Task를 선택하고 실행한다.
Webserver
Scheduler에서 분석한 DAG를 시각화하고 DAG 실행과 결과를 확인할 수 있는 인터페이스를 제공한다. 인터페이스로 보여지는 정보들은 Metadata Database에서 가져온다.
Metadata Database
Scheduler, Worker, Webserver의 상태정보를 저장한다.
Airflow는 어떻게 동작하는가?
- DAG로 워크플로를 작성한다. DAG가 모여있는 DAG Directory 폴더에 추가된다.
- DAG Directory에 있는 추가된 DAG를 Scheduler가 분석하여 예약 주기가 현재 시간 이전이라면 Task 실행을 예약한다.
- Task의 의존성이 모두 해결되었다면(? 책에 나와있는 설명이 다름..12, 13 페이지 참조) 실행 대기열에 Task를 추가한다.
- Worker들 중 pool worker가 실행 대기열에 있는 Task를 선택하고 실행한다. (이 때, 실행은 병렬로 처리된다.)
- 이 과정들의 결과가 Metastore로 전달된다.
- Webserver는 Metastore의 정보를 인터페이스에 반영하고, 이를 통해 Task 실행과 결과를 모니터링한다.
Airflow가 적합하지 않은 경우
Airflow는 반복적이거나 배치 Task를 실행하는 기능에 특화되어있기 때문에 실시간 데이터 처리를 하는 스트리밍 워크플로에는 적합하지 않다.
또한 Airflow는 Python 함수를 통해 동적 Task를 생성할 수 있지만 DAG에는 이러한 동적 Task가 표현되지 않아 Task의 추가/삭제가 빈번한 동적 파이프라인의 경우에는 적합하지 않을 수 있다.
