2년 동안 Kubernetes에서 Airflow를 실행하고 배운 것 (번역)

medium에서 What we learned after running Airflow on Kubernetes for 2 year 을 읽으며 요약한 내용입니다.
 
이 게시글에서는 Apache Airflow를 사용하여 데이터 플랫폼을 구축하는 과정에서 경험한 내용들을 공유하고 있습니다. 저자는 Airflow를 사용하여 데이터 변환, 사기 탐지, 데이터 과학 이니셔티브, 내부 작업 등을 관리합니다. 현재 300개 이상의 DAG가 production에 존재하며, 하루에 평균 5,000개 이상의 작업을 실행합니다. 이러한 규모의 작업에도 8개월간 실패 없이 진행되었다고 합니다. 
 
production level에서 Airflow를 사용하며 얻게 된 아이디어들을 공유하고 있고 요약하자면 아래와 같습니다. 

1. Executor 선택: 처음에는 Kubernetes Executor를 선택했지만 작은 작업에 대한 오버헤드와 Pod 강제 이탈로 인한 문제가 발생했습니다. 이후 Celery Executor로 전환하여 문제를 해결하고 성능을 향상시켰습니다.

2. DAG 동적 생성과 디커플링: 여러 팀이 자체 DAG를 작성하고 이를 통합하기 위해 DAG의 멀티-저장소 접근 방식을 도입했습니다. DAG는 개별 리포지토리에서 개발되며, CI/CD 가이드라인을 사용합니다.

3. 세부 설정 조정: CeleryExecutor로 전환한 후 메모리 누수와 같은 새로운 문제가 발생했습니다. 이를 해결하기 위해 worker_max_tasks_per_child 및 worker_max_memory_per_child와 같은 Celery 구성을 조정했습니다.

4. 알림, 경고 및 관측성: Airflow의 알림 및 경고 시스템을 개선하기 위해 Slack 메시지 및 Prometheus 규칙을 설정했습니다. 또한, 에어플로우 및 인프라 모니터링을 위해 메트릭 및 로그를 수집하고 사용했습니다.

5. 비용 최적화: 작업의 동시성 및 리소스 구성을 조정하여 비용을 최적화했습니다.

 
각각의 단계별로 살펴보면 아래와 같습니다.

1. Executor 선택

처음에 저자는 Kubernetes 환경에서 구축하여 사용하기 때문에 Kubernetes Executor 를 사용했습니다. 하지만 운영되는 task 들은 경량의 DBT 변환 작업이고 적은 수의 모델만이 긴 시간(+/- 1 시간)을 소요합니다. 이들은 두가지 문제에 직면했는데요.
- Kubernetes Excutor를 사용시 작업 시작 오버헤드 존재. pod 시작 대기 시간이 작업 자체 시간보다 길 때가 있음.
- pod 제거로 일부 작업(특히 장기 실행 작업)이 예기치 않게 실패함.
 

신속하게 스케일이 조절되는 k8s의 특성으로 문제가 발생

 
 
결국 저자는 Celery Executor로 돌아가는 결정을 했습니다. worker nodes를 고정하여 작고 규모의 작업이 많은 사용 사례에 적합하게 활용할 수 있었습니다. 
 

2. 분리 및 동적 DAG 생성

분산된 DAG 저장소

모든 DAG는 DAG를 소유하는 팀에 따라 특정 경로에서 동기화 프로세스를 통해 버킷에 저장합니다.

->  CI/CD가 매우 중요하며 각 DAG 이름은 DAG ID가 충돌하지 않도록 미리 지정된 접두어로 시작
-> 정적 검사(올바른 소유자를 할당하고 태그의 존재여부 확인)
-> 모든 DAG는 커밋되기 전 최소한의 체크리스트를 통과해야 합니다.

DAG를 에어플로우에 동기화하는 방법

버킷의 내용을 스케줄러, 워커 등의 Pod의 로컬 파일 시스템에 동기화해야 합니다.
저자는 Objinsync 를 사용하여 원격 객체 저장소를 로컬 파일 시스템으로 점진적으로 동기화하는 가벼운 데몬을 사용했습니다.
 
메인 스케줄러나 워커 컨테이너가 시작되기 전에 objinsync를 init 컨테이너로 추가해야 합니다.(특히 Celery 워커의 경우에 중요하게 작용)

동적으로 DAG 생성 시 주의사항

대규모로 DAG를 생성하려면 DAG 템플릿화와 프로그래밍 방식의 생성을 활용해야 합니다. 더 이상 모든 DAG를 수동으로 작성하지 않습니다 😂

동적으로 DAG를 생성하는 가장 간단한 방법은 단일 파일 방법을 사용하는 것입니다. (동적으로 생성하려는 각 DAG에 대해 .py 파일을 생성합니다.)
 
동적으로 생성된 파일을 DAG 버킷에 푸시합니다.
단일파일 / 다중파일으로 동적으로 DAG 생성하는 방법은 여기를 참고하세요

3. 설정 세부 조정

CeleryExecutor로 전환하자마자 일부 Celery 워커가 OOM(메모리 부족) 문제로 인해 다운되기 시작했습니다. 

Celery 워커의 메모리 누수 발견!

메모리 누수를 방지하고 작업의 메모리 사용량을 제어하기 위해 두 가지 중요한 Celery 구성을 조정해야 했습니다: `worker_max_tasks_per_child` 및 `worker_max_memory_per_child`.

`worker_max_tasks_per_child`: 첫 번째 구성은 워커 프로세스가 새 프로세스로 교체되기 전에 실행할 수 있는 최대 작업 수를 제어합니다. 먼저 Celery 워커와 워커 프로세스의 차이를 이해해야 합니다. 하나의 워커 노드는 concurrency 설정에 의해 제어되는 여러 워커 프로세스를 생성할 수 있습니다. 예를 들어, concurrency를 12로 설정하고 2개의 Celery 워커가 있다면 24개의 워커 프로세스가 생성됩니다.

따라서 동일한 워커 프로세스 내에서 작업 사이의 메모리 누수를 피하기 위해 때때로 리사이클하는 것이 좋습니다. 이 구성이 설정되지 않으면 워커 프로세스를 리사이클하지 않는 것이 기본 설정입니다.

`worker_max_memory_per_child` 두 번째 구성은 단일 워커 프로세스가 새로운 프로세스로 교체되기 전에 실행할 수 있는 최대 주요 메모리 양을 제어합니다. 이것은 사실상 작업의 메모리 사용량을 제어합니다. 기본값도 제한이 없으므로 항상 이를 설정하는 것이 좋습니다.

이러한 두 구성을 조정함으로써 최대 작업 수에 도달하거나 최대 주요 메모리 양에 도달한 경우 워커 프로세스를 리사이클하여 메모리 사용량을 제한합니다. 이러한 구성이 작동하는 것은 prefork 풀을 사용하는 경우에만 해당됩니다. 공식 문서에서 자세히 알아보십시오.

Airflow에서 이를 설정하는 것은 매우 간단합니다. Airflow의 config_templates 폴더에서 기본 Celery 구성을 업데이트하기만 하면 됩니다
 

# config_templates/custom_celery.py

from airflow.config_templates.default_celery import DEFAULT_CELERY_CONFIG

CUSTOM_CELERY_CONFIG = DEFAULT_CELERY_CONFIG.copy()
CUSTOM_CELERY_CONFIG.update(
    {
        "worker_max_tasks_per_child": <int>,
        "worker_max_memory_per_child": <int>,
    }
)

그런 다음 `values.yaml`에서 이 사용자 지정 구성을 가리키면 됩니다.

airflow:
   config:
     celery:
       worker_concurrency: <int>
       celery_config_options: config_templates.custom_celery.CUSTOM_CELERY_CONFIG

이러한 구성에 사용하는 값은 작업 노드 구성, 메모리 요청/제한 양, concurrency 수준 및 작업의 메모리 집중도에 따라 달라집니다. 따라서 개별 설정에 맞게 조정해야 합니다.

노드 로테이션에 대비하기

노드 로테이션으로 Pod가 종료될 때 Pod가 비활성화되어 실패하는 것을 방지하기 위해 개별적인 요구 사항에 따라 Worker Termination Grace Period 구성을 설정하는 것이 중요합니다.
이 구성은 릴리스 프로세스나 노드 로테이션에 의해 종료되기 전에 celery worker가 해당 시간 동안 대기하도록 만듭니다. 이것은 Airflow의 차트 values.yaml에서도 쉽게 설정할 수 있습니다.

airflow:
   workers:
      terminationGracePeriodSeconds: <int>

  
권장하는 방법은 가장 긴 실행 작업에 소요된 시간의 1.5배로 설정하는 것입니다. 따라서 모든 작업이 해당 기간 내에 완료될 수 있고 워커가 원활하게 종료될 수 있습니다.

4. 알림, 경고 및 관측성

조직 내 알림 통합

여러 팀이 Airflow를 사용하는 환경에서 작업 중인 경우, 알림 메커니즘을 통합해야 합니다.
Airflow 2.6부터는 Notifiers를 사용하여 이를 관리할 수 있습니다. 사전 구축된 알림을 사용할 수도 있습니다! (Slack, SQS, Jira 등)
BaseNotifier 클래스를 이용하면 사용자 정의 Notifier를 만들 수 있습니다. 

한 가지 알림, 여러 대상 및 사용자 정의

사용자 지정 알림도 템플릿화되어 팀이 Slack에서 정보 메시지를 만들기 위해 표준 형식을 사용할 수 있습니다. 이 접근 방식의 또 다른 장점은 이를 사용하는 개별 팀이 개별 알림 대상의 비밀 관리에 대해 걱정할 필요가 없다는 것입니다.

실패에 대한 최초의 알림

고가용성을 고려해서 구현하더라도 Airflow는 여러 이유로 여전히 실패할 수 있습니다. 그래서 인프라 수준의 관측성, 메트릭 및 경고가 매우 중요합니다.

Kubernetes에서 실행 중일 때, 관심 있는 각 이벤트에 대한 PrometheusRule을 설정하여 수행할 수 있습니다. 예를 들어, 스케줄러 노드의 상태, 사용 가능한 워커 노드의 수 또는 스케줄러 루프 시간과 같은 특정 Airflow 메트릭을 모니터링할 수 있습니다. 또한 AlertManager를 실행함으로써 여러 대상에 대한 경고를 발생시킬 수 있습니다(Slack, PagerDuty, Opsgenie 등).

환경의 관측성을 향상시키기 위해 Airflow 메트릭을 활용하는 것도 좋은 방법입니다. 현재 Airflow는 `StatsD`와 `OpenTelemetry`로 메트릭을 보내는 것을 지원합니다. 후자가 전자보다 우선되는데, `OpenTelemetry`는 로그 및 트레이스를 지원하는 더 완벽한 프레임워크이기 때문입니다. 

또한, 이를 사용하려면 Kubernetes에서 OTEL Collector 배포를 관리해야 합니다(공식 helm 차트 참고). 공식 Airflow 차트는 OTEL Collector를 제공하지 않습니다.

Airflow 모니터링 차트 예시

표준 메트릭을 사용하면 알림의 품질을 향상할 수 있습니다. 예를 들어, 대기 중인 작업의 총 수를 사용하여 대기열이 특정 시간 동안 너무 많이 증가할 때 트리거할 경고를 설정할 수 있습니다. 예를 들어 SLA 시간보다 긴 대기열은 원하지 않습니다.

우리가 모니터링하는 다른 유용한 메트릭에는 DAG 구문 분석 시간 및 스케줄러 루프 시간이 포함됩니다. 이를 통해 Airflow의 핵심에 영향을 미치고 전체 응용 프로그램을 느리게 만드는 문제를 빠르게 식별할 수 있습니다.

Airflow의 메타데이터 관리하기

메타데이터 데이터베이스는 성공적인 Airflow 구현의 중요한 부분으로, 성능을 저하시키거나 Airflow를 중단시킬 수 있습니다.

위에서 언급한 Airflow 노드 및 성능 메트릭의 모니터링 외에도 데이터베이스 상태 메트릭을 모니터링하는 것이 중요합니다. PostgreSQL 또는 MySQL을 사용하고 있는지에 따라 다를 수 있지만(제발 SQLite를 사용하지 마십시오 😂), 가장 일반적인 메트릭은 CPU 사용량, 사용 가능한 저장소, 연결 수 등입니다.

또 다른 좋은 실천 방법은 주기적으로 오래되고 사용되지 않는 메타데이터를 정리하는 것입니다. 이에는 작업, `dag_run` , `task_instance`, `log`, `xcom, sla_miss`, `dags`, `task_reschedule`, `task_fail`등과 같은 테이블이 포함될 수 있습니다. 이 모든 메타데이터는 Airflow 내에서 계속해서 쌓이며, 작업 상태를 검색하는 평균 쿼리가 더 길어질 수 있습니다. Airflow가 단순히 로드되고 탐색하는 데 느린 것으로 느껴진 적이 있다면 메타데이터가 쌓이는 것이 그 원인일 수 있습니다.

Airflow에는 메타데이터 정리를 위한 네이티브 명령인 airflow db clean이 있으며, 추가적인 옵션은 여기를 확인하세요.

Kubernetes를 사용하는 경우, 이를 위한 CronJob을 Airflow 차트에 추가 리소스로 설정하여 지정한 플래그와 함께 주기적으로 airflow db clean 명령을 실행하는 것이 좋은 방법입니다. 일일 또는 주간으로 실행해야 할 수 있습니다.