[GCP] Composer 설명

GCP Composer 란?

GCP Composer는 워크플로우 자동화를 위한 웹 서비스로, Apache Airflow를 기반으로 구축되었다. 결국 파이프 라인을 구축하고 관리하는 웹 서비스라고 생각하면 된다 비슷한 용도는 jenkins, airflow, argo workflow 등이 있다. python 기반으로 작성된 코드를 통해 워크플로우를 구축하고 관리할 수 있고 GCP를 이용 한다면 좋은 선택이 될 수 있을 것이다.

airflow architecture

• DAG는 워크플로우를 구성하는 코드이다.

• TASK는 DAG에서 정의된 각각의 작업 단위이다.

• Scheduler는 워크플로우를 예약하고 실행하는 역할을 한다.

• Worker는 실제로 워크플로우를 실행하는 cpu 와 메모리가 할당 된 서버 같은 개념이라고 보면 된다.

• Executor는 워크플로우를 실행하는 역할을 한다.

alt text

airflow scheduler 실행 흐름

일꾼이 아무리 많아도 일을 분배 해주는 작업반장이 없다면 같은 일을 두번 처리한다거나 효율적인 일의 배분이 이루어지지 않을 것이다.

수단과 방법을 가리지 말고 DAG를 airflow에 올려 놓으면 지정 된 시간에 batch 형태로 실행 된다. 아래 그림을 보면 queue에 들어가고 있는 모습을 볼 수 있고 보통 저 솔루션은 redis를 이용하여 메모리에 저장하고 있다가 실행된다. 저 queue의 존재로 인해 여러가지 작업을 여러 worker가 동시에 실행 할 수 있는 것이다.

alt tet

GKE(Google Kubernetes Engine) architecture

쿠버네티스는 왜 알아야 하는건가요?

composer와 쿠버네티스(GKE)는 밀접한 관계가 있다.

저 위에 있는 redis나 ui를 제공해주는 webserver 등의 서비스은 모두 k8s의 pod로 구성 되어 있고 클러스터 내부에서 서로 유기적으로 연결 되어 있다. worker는 지정 된 시간에 k8s batch의 형태로 pod로 실행 될 것이다. 그 외에 설정 파일은 k8s의 configmap으로 구성 되어 있고 보안 파일은 k8s의 secret으로 구성 되어 있다.

alt text

gcp composer architecture

composer v2 기준으로 왼쪽 gke 내부에 구성 되어 있고 tenant라고 되어 있는 부분이 GCP에서 자체 체공 하는 서비스인데 조금 특이한 점은 이 영역에 구성된 솔루션들은 직접 접속을 하거나 내부를 들여다 볼수 없는 serverless형태로 구성 되어 있다. 버전이 올라 갈수록 점점 serverless 형태로 구성 되어 있는 것을 볼 수 있다.

alt text

GCP Composer 모니터링 지표

위에 설명한 architecture를 이해 하였다면 모니터링 지표가 어느정도 이해가 될 것이다.

Metric	Filter	이슈발생
Environment health	Healthy	composer health check
SQL Database health	Database Healthy	airflow database health check
CPU utilization per node	CPU usage time	k8s Node에서 사용된 누적 CPU 사용량
Memory Utilization per node	Memory usage	k8s Node에서 사용한 누적 메모리
Database Memory usage	Database memory quota	데이터베이스 인스턴스의 최대 RAM 크기
	Database memory usage	데이터베이스 인스턴스의 메모리 사용량
Kubernetes Container	CPU usage time	k8s 컨테이너에서 사용하는 모든 코어의 누적 CPU 사용량
Kubernetes Container	Limit cores	k8s 컨테이너의 CPU 제한
Airlfow Active workers	Celery Workers	celery worker의 수
Airflow Worker Pod evictions	Worker Pod Eviction Count	worker pod 강제 종료 횟수
Airflow scheduler hearbeat	Scheduler Heartbeats	scheduler heatbeat
Dag Bag Size	Dag Bag Size	dag bag size 사이즈
Task Instance Count	Task Instance Count	task instance 처리 현황
Task Queue Length	Task Queue Length	task queue 수
Dag parsing processes	DAG parsing processes	현재 실행 중인 DAG parsing 프로세스 수
Executor Running Tasks	Executor Running Tasks	실행 Tasks 수
Dag Parsing Error Count	Parse Error Count	DAG Parsing Error수
Total Parse Time	Total Parse Time	모든 DAG를 SCAN하고 IMPORT하는데 소요되는 시간
Unfinished Task	Unfinished Task Instances	완료되지 않은 전체 task 인스턴스
Workflow Runs	Workflow Runs	workflow 수
Workflow Run Duration	Workflow Run Duration	workflow 수행시간
Tasks	Tasks	operator별 실행 건수
Task Duration	Task Duration	operator별 실행 시간

추가 용어 설명

dagbag_size

dagbag_size는 Airflow에서 DAGBag이 스캔할 때 발견한 DAG의 수를 나타내는 메트릭입니다. 이 메트릭은 스케줄러가 DAG 디렉토리를 스캔할 때마다 업데이트됩니다. 이를 통해 현재 시스템에 로드된 DAG의 총 수를 모니터링할 수 있습니다. 이 메트릭은 다음과 같은 상황에서 유용할 수 있습니다: DAG의 수가 예상치와 일치하는지 확인할 때, DAG 디렉토리의 변경 사항을 모니터링할 때 시스템의 부하를 관리할 때

Airflow에서 workflow, DAG, task, run의 차이를 이해하는 것은 중요합니다.

1. Workflow: 정의: Workflow는 일련의 작업을 정의한 것입니다. Airflow에서는 DAG로 표현됩니다. 설명: Workflow는 여러 작업(task)들이 특정 순서와 조건에 따라 실행되는 전체적인 흐름을 의미합니다.

2. DAG (Directed Acyclic Graph): 정의: DAG는 작업 간의 의존성을 정의한 그래프입니다. 설명: DAG는 작업들이 어떤 순서로 실행되어야 하는지를 정의합니다. 각 작업은 노드로, 작업 간의 의존성은 엣지로 표현됩니다. DAG는 순환이 없어야 합니다.

• Task:

  • 정의: Task는 DAG 내에서 실행되는 개별 작업입니다.

  • 설명: Task는 특정 작업을 수행하는 단위입니다. 예를 들어, 데이터베이스에서 데이터를 추출하거나, 데이터를 처리하는 작업 등이 있습니다.

1. Run:

• 정의: Run은 DAG 또는 Task가 실제로 실행되는 인스턴스를 의미합니다.

• 설명: DAGRun은 특정 시간에 DAG가 실행된 인스턴스를 의미하며, TaskInstance는 특정 시간에 Task가 실행된 인스턴스를 의미합니다.

• 예시: 매일 자정에 실행되는 DAG가 있다면, 매일 하나의 DAGRun이 생성됩니다. 각 DAGRun 내에서 여러 TaskInstance가 생성되어 실행됩니다.