airflow 파라미터 튜닝

앞서 docker-compose를 통해 airflow를 심플하게 설치하는 방법에 대해서 다뤄봤습니다. 스터디, 혹은 가벼운 인프라에서 사용되는 목적이라면 기본 설치된 airflow 환경으로 충분합니다. 하지만 조금 더 묵직한 환경에서 운영이 되어야 한다면 튜닝이 필수인데요, 튜닝에 어떤 것들이 있는지 살펴보도록 하겠습니다.

---

# concurrency, max_active_runs

아마도 튜닝의 첫 단계가 아닐까 싶습니다. airflow의 목적은 병렬로 다수의 task를 운영하는 것이기 때문에 concurrency, max_active_runs 옵션은 제일 먼저 손봐줘야 하는 부분이 될 겁니다. 각각이 나타내는 의미는 아래 그림에서 가장 잘 표현되어 있습니다. 이 이상의 설명은 존재할 수 없을 듯.

https://stackoverflow.com/a/63955004/4599185

설정은 아래와 같이 DAG에서 해주면 됩니다. concurrency가 10이므로 hello_world_dag이 몇 개이든 task는 10개까지만 실행됩니다. 한편 max_active_runs이 2로 설정되어 있기 때문에 dag은 최대 2개까지만 동시 실행됩니다. 더 구체적으로 이야기하면 hello_world_dag 안에 7개의 task가 있다고 가정했을 때 아래와 같은 설정이라면 2개의 dag이 실행되지만 첫 번째 dag에서 5개의 task가 실행되므로 두 번째 dag은 2개의 task만 실행이 가능합니다. 앞에 5개의 task가 종료됨에 따라 두 번째 dag의 나머지 3개 task가 실행되게 됩니다. ( 말이 좀 어렵죠? 위에 그림을 보면 더 쉽게 이해가 될 겁니다 ) 

with DAG(dag_id="hello_world_dag",
     default_args=default_args,
     concurrency=10,
     max_active_runs=2,
     catchup=False,
     schedule_interval='* * * * *',
     start_date=datetime(2021, 7, 5, tzinfo=KST),
     tags=['hello', 'world'],
) as dag:

 

# parallelism

조직 내에 airflow의 성장세가 가속됨에 따라 동시에 실행되어야 하는 dag의 개수가 엄청나게 많아졌다고 가정합시다. 안타깝게도 airflow는 기본 설정으로는 최대 32개의 task만 동시 실행이 가능합니다. 이는 parallelism 설정 때문인데요. 32개 이상을 실행시키기에는 기본 환경에서 무리가 있다고 판단되기 때문이죠. 여기서 말하는 기본 환경은 단일 컴퓨팅 환경입니다. 아무리 그래도 그렇죠, 요즘 컴퓨팅 파워가 얼마나 좋아졌는데 32개는 조금 너무했네요. airflow의 parallelism은 기본 값이 32지만 당연히 변경할 수 있습니다. docker-compose로 설치가 됐다면 docker-compose.yaml 파일 environment에 AIRFLOW__CORE__PARALLELISM을 추가하고 원하는 값을 입력해주세요.

version: '3'
x-airflow-common:
  &airflow-common
  image: ${AIRFLOW_IMAGE_NAME:-apache/airflow:2.1.0}
  environment:
    &airflow-common-env
    AIRFLOW__CORE__EXECUTOR: CeleryExecutor
    AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@postgres/airflow
    AIRFLOW__CELERY__BROKER_URL: redis://:@redis:6379/0
    AIRFLOW__CORE__FERNET_KEY: ''
    AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 'true'
    AIRFLOW__CORE__LOAD_EXAMPLES: 'true'
    AIRFLOW__API__AUTH_BACKEND: 'airflow.api.auth.backend.basic_auth'
    AIRFLOW__CORE__PARALLELISM: '256'
    AIRFLOW__CELERY__WORKER_CONCURRENCY: 256
  volumes:
    - ./dags:/opt/airflow/dags

추가적으로 AIRFLOW__CELERY__WORKER_CONCURRENCY도 설정해줘야 합니다. parallelism을 수정한다고 해도 실제 task를 실행시키는 celery의 큐를 늘려주지 않으면 소용이 없기 때문입니다. celery의 max concurrency 기본 값은 16으로 설정되어 있습니다. 5555번 포트로 접속해서 Pool size를 조정해도 되지만 컨테이너가 재시작되면 초기화되기 때문에 docker-compose.yaml 파일에 직접 수정하는 것을 권장합니다.

 모든 수정이 끝나면 아래 명령어를 통해 airflow 환경을 재시작해줍니다. 재시작하게 되면 docker-compose.yaml에 수정된 내용이 읽혀서 환경에 반영됩니다.

docker-compose down && docker-compose up -d

변경사항이 잘 적용되었는지 의심스럽다면 container 중에 하나를 들어가서 아래 명령어를 통해 확인하시면 됩니다.

airflow config get-value core parallelism

 

# postgres max_connections

대량의 task를 실행하기 위한 최종 관문입니다. airflow 환경에서 사용하는 postgres의 설정을 수정해줘야 합니다. postgres의 기본 max_connections은 100이기 때문에 그것보다 많은 task가 동시에 실행된다면 connection을 맺지 못하고 오류가 발생하게 됩니다. postgres container로 접속해서 설정을 해줍시다. 우선 아래처럼 컨테이너로 접속하고요. 

docker exec -it 9370bb8afef4 /bin/bash

psql로 접속해줍니다.

root@9370bb8afef4:/# psql -U airflow
psql (13.3 (Debian 13.3-1.pgdg100+1))
Type "help" for help.

airflow=#

그리고 아래 쿼리를 날려줍니다.

ALTER SYSTEM SET max_connections TO '500';

그리고 컨테이너를 빠져나와서 postgres를 재시작해줍니다. 설정된 값은 재시작해야 적용됩니다. 

 

# 마무리

사실 튜닝의 최종단계는 airflow executor를 변경하는 것입니다. 기본 값인 CeleryExecutor의 경우 단일 머신에서 동작하기 때문에 튜닝을 해도 동작에 제약이 있을 수밖에 없습니다. 그래서 찾아온 executor가 KubernetesExecutor입니다. airflow 1.10.0 버전부터 Kubernetes Executor를 지원하기 시작했는데 pod를 통해 task를 실행시키고 인프라 자원을 효율적으로 관리할 수 있게 됐습니다. KubernetesExecutor를 사용하는 경우 아래와 같은 구조를 갖게 되는데 이때는 redis, flower가 필요 없어집니다. 다만, kubernetes 인프라를 관리해야 할 필요가 생깁니다.

KubernetesExecutor 구조 (https://airflow.apache.org/docs/apache-airflow/stable/executor/kubernetes.html)

KubernetesExecutor에 대한 이야기는 별개의 글을 통해 다뤄보는 시간을 갖도록 하겠습니다. CeleryExecutor를 벗어날 고민을 하게 될 때쯤엔 조직에서 사용하는 airflow가 상당히 무르익었다고 생각할 수 있기 때문에 아주 신나게 교체를 진행할 수 있을 겁니다. :-)