[Hands On] 아파치 에어플로우 도커환경에서 핸즈온 - #1 기초 (Apache Airflow)

아파치 에어플로우는 '작업흐름'을 코드를 통해 작성하고, 스케쥴하고, 모니터링하기 위한 데이터 워크플로우 매니지먼트 도구입니다. 여기서 작업흐름은 하나이 쉘 스크립트를 실행하는 것, 웹사이트에 접근에 특정 수행을 통해 데이터를 다운 받고 다운이 받아졌다면 파싱하여 슬랙 알림을 보내는 것 등 다양한 의존적인 작업들의 흐름이 포함됩니다. 

 

Tasks in Apache Airflow DAG - Image from Apache Airflow Doc

이 글에서는 도커 환경에서 아파치 에어플로우 클러스터를 빠르게 올려서 아래와 같은 사항들을 중점으로 설명드리도록 하겠습니다:

 

• 준비사항
• 주요개념
• 간단한 작업 수행 시 일어나는 일들
• 아파치 에어플로우 클러스터 아키텍쳐

---

준비사항 및 클러스터 띄우기

아래 도커 환경에서 아파치 에어플로우를 실행하기 위해서는 아래와 같은 것들이 필요합니다:

 

• Git
• Docker 및 Docker Compose
• 기본적인 Python에 대한 이해

위와 같은 부분들이 준비되었다면, 아래 코드를 실행해서 에어플로우 클러스터를 띄울 수 있습니다:

# docker-compose 파일 클론
git clone https://github.com/kadensungbincho/de-hands-on.git
cd de-hands-on/docker-apache-airflow-poc

# 기본적인 폴더 생성
mkdir ./dags ./logs ./plugins

# 에어플로우 초기 셋업 및 Redis와 Postgres 시작
docker-compose up airflow-init

# 에어플로우 클러스터 시작
docker-compose up -d

 

아래와 같이 현재 실행되는 컨테이너 리스트를 확인하여 에어플로우 클러스터가 잘 생성되었는지 살펴볼 수 있습니다:

docker container ls

# output
... Names
... docker-apache-airflow-poc_flower_1
... docker-apache-airflow-poc_airflow-worker_1
... docker-apache-airflow-poc_airflow-webserver_1
... docker-apache-airflow-poc_airflow-scheduler_1
... docker-apache-airflow-poc_airflow-init_1
... docker-apache-airflow-poc_postgres_1
... docker-apache-airflow-poc_redis_1

 

또한, http://localhost:8080을 (아이디 / 패스워드는 airflow / airflow) 통해 아래와 같은 UI를 확인할 수 있습니다:

Apache Airflow Main Page - Image from Author

 

 

주요개념

일상생활에서 하나의 작업을 진행하는 작업순서도가 있다고 하면, 아래와 같은 찌개끓이기가 그러한 작업순서도의 하나라고 할 수 있습니다. 

A workflow of making a Soy Soup - Image from Author

손씻기, 재료손질 등의 각 단계로 이루어진 작업순서도는 '한 방향'으로만 흐르며, 각 단계는 이전에 다른 단계가 진행되어야 진행할 수 있는 의존성을 가지기도 합니다.

 

에어플로우도 컴퓨터로 실행하는 다양한 위와 같은 '작업순서도'를 코드로 구현하고 스케쥴을 정해서 자동으로 실행될 수 있도록 도와주는 도구입니다. 그렇기에 그러한 작업흐름의 각 요소들을 일컫는 다양한 개념들을 만들어 두었고, 그런 부분이 에어플로우 코드로 구현되어 있습니다.

 

• DAG (Directed Acyclic Graph): DAG은 하나의 작업흐름 및 실행 정보를 가지며, 이러한 DAG의 실행들은 DAG_RUN으로 관리됩니다.
• Task: 하나의 task는 하나의 DAG 안의 작업 단위를 말하며, DAG에서 각 노드에 해당됩니다. 실행 시 DAG -> DAG_RUN으로 인스턴스가 관리되는 것과 유사하게 task 역시도 Task Instance로 각 task의 실행이 관리됩니다. 
• Operator: 오퍼레이터는 특정 작업을 실행하기 위한 템플릿으로, DAG의 각 task들은 목적에 맞는 Operator를 사용하여 이루어 집니다. 

DAG, DAG_RUN, Task, Task Instance - Image from Author

 

간단한 작업 수행 시 일어나는 일들

먼저, 간단한 작업을 하나 수행하며 동작방식을 살펴보겠습니다. 

 

에어플로우에서 하나의 작업은 task라는 이름으로 불립니다. 이러한 task들의 묶음은 DAG (Directed Acyclic Graph)를 단위로 관리되는데요. 이 DAG은 task들 간의 관계나, 의존성을 반영하여 구조화되어 있습니다.

 

에어플로우 첫 화면에서 보이는 것은 DAGs라는 글자가 크게 쓰인 아래와 같은 페이지로 example_*이라는 이름을 가진 DAG들이 나열되어 있는 것을 볼 수 있습니다. 각 행이 하나의 DAG로 화면을 통해서도 DAG과 관련해 다음과 같은 정보들이 존재한다는 부분을 알 수 있습니다:

 

Image from Author

• 이름
• 소유자
• DAG 실행결과 상태에 따른 Count (성공, 중지, 실패한 DAG Run의 합계)
• 스케쥴: 어떤 스케쥴로 하나의 DAG이 실행될지
• 최종 수행일
• 최근의 Tasks 상태에 따른 Count
• DAG에 대한 액션들 (실행, Refresh, 삭제)

 

위와 같이 하나의 DAG은 'task들의 묶음이 어떻게 수행되기 원하는지?'에 대한 정보를 가지며 하나의 작업흐름을 나타냅니다.

 

위에서 'On' 상태로 변경한 DAG을 실행(Trigger)해보겠습니다. 오른쪽의 'Trigger DAG' 버튼을 누릅니다:

 

 

아래와 같은 페이지가 나오면, (현재는 특별한 conf가 없으므로) Trigger를 눌러줍니다:

 

그러면 아래와 같이 연두색으로 보이는 Running 상태의 동그라미에 1이 나타난 것을 볼 수 있습니다:

 

조금 더 기다린 후 Refresh하면 success 상태의 동그라미에 1이 표시되고, DAG_RUN이 성공적으로 끝난 것을 볼 수 있습니다:

 

DAG 이름인 example_python_operator를 클릭하면 DAG이나 DAG_RUN의 상세 페이지로 이동할 수 있습니다. 한 번 더 'Trigger DAG'을 눌러주고, DAG 이름을 클릭하여 이동하면 아래와 같은 Tree View로 DAG_RUN에 대한 정보를 보여줍니다:

 

하나의 네모칸은 위에서 살펴본 것과 같이 DAG_RUN을 이루는 노드에 해당되는 Task Instance인 것을 알 수 있습니다. 네모칸에 마우스 커서를 hover하면 하나의 Task Instance가 id는 무엇인고, 어떠한 DAG_RUN에 속하는지, 어떠한 오퍼레이터를 사용하는지, 실행 시작 시간은 언제고 얼마 동안 실행되었는지 등을 보여주고 있습니다:

 

해당 네모칸을 클릭하면 아래와 같은 창이 뜹니다:

위의 빈 파란색 테두리의 버튼들을 통해 해당 Task Instance에 대한 상세 정보를 살펴볼 수 있습니다. 아래 Task Actions를 통해 Task Instance의 의존성을 변경하고 실행하거나, 상태를 변경할 수 있습니다 (Clear 시에 실행이 안된 것으로 간주되어 한 번 더 실행됩니다).

 

아래에서 print_the_content라는 id를 가진 Task Instance를 Downstream, Recursive 버튼이 선택된 아래와 같은 상황에서 Clear를 눌러보겠습니다:

 

누르면 실수를 방지하기 위해 아래와 같은 추가적인 확인 페이지가 뜨고 다시 OK를 누르면,

 

아래와 같이 print_the_context에 의존하였던 Task Instance들도 Clear되어 한 번 더 실행되는 것을 볼 수 있습니다:

 

이후 print_the_context Task Instance 네모칸을 누르고 아래와 같이 Instance Details 버튼을 눌러 Task Instance 상세 정보를 살펴보겠습니다:

다른 것보다도 Task Instance가 잘 실행되었는지, 실패했다면 원인은 무엇인지 살펴볼 때에 아래와 같이 Task Instance의 Log로 들어가서 Log를 확인해서(또는 코드 상에 로그를 추가하고 이곳에서 확인해서) 빠르게 파악할 수 있습니다:

 

 

이어 가장 오른쪽의 Code <> 버튼을 누르면 파이썬으로 작성된 DAG의 코드를 확인할 수 있습니다:

 

이제까지 간단한 DAG을 실행하여, DAG_RUN을 생성하고, 예하의 여러 Task Instance들이 생성되어 작업이 실행되는 것을 살펴보았습니다.

 

다음 장에서는 감을 잡은 위의 Flow를 만들기 위해 에어플로우가 어떤 아키텍쳐로 구성되어 있는지 살펴보며 추후 복잡한 사항이 발생해도 어디를 확인해야 하는지 알 수 있도록 전달해보겠습니다.

 

 

아파치 에어플로우 클러스터 아키텍쳐

위에서 docker-compose up을 통해 컨테이너들을 생성하면 아래와 같은 형태로 컨테이너(와 내부의 서비스)가 실행됩니다:

 when you typed docker-compose up - Image from Author

 

이미지에서 가장 왼쪽부터 시계방향으로 살펴보면,

 

• Airflow init: airflow의 기본적인 사항들을 설정합니다 (예, Postgres의 Airflow 테이블 등)
• Postgres: 에어플로우의 다양한 객체들의 상태를 저장합니다.
• Redis: Airflow Scheduler가 다양한 task instance를 저장하며, 그러한 task instance들은 Flower가 Worker 노드에 할당하여 처리하게 됩니다.
• Flower: Airflow Worker에 Redis에 쌓인 task instance들을 할당합니다.
• Airflow Worker: 실제 task를 수행합니다.
• Airflow Scheduler: DAG_RUN의 task instance들을 스케쥴링합니다.
• Airflow Webserver: Airflow의 Web UI로 사용자들의 권한에 따라 다양한 액션을 GUI 환경에서 수행할 수 있습니다.

 

위의 다양한 컨테이너 안의 서비스들은 UI 또는 Client를 통해 확인할 수 있습니다.

 

Postgres에 airflow 데이터베이스의 내용을 살펴보면, 다음과 같은 다양한 테이블이 생성되어 있는 것을 볼 수 있습니다:

airflow database tables - Image from Author

Airflow Webserver는 이전에 살펴본 것처럼, 8080 포트를 통해 확인할 수 있습니다.

 

Flower 또한 GUI를 제공하기에, http://localhost:5555로 접속하여 아래와 같은 Tasks들을 확인할 수 있습니다:

 

 

Reference

[1] hub.docker.com/r/apache/airflow

[2] Apache Airflow Official Dockerfile

[3] airflow.apache.org/docs/apache-airflow/2.0.1/start/docker.html