Airflow DAG 실행 시간, 이렇게 줄였습니다: 실전 최적화 경험기

Airflow DAG 실행 시간이 길어 고민이라면? 태스크 의존성 최적화와 병렬 처리 기법으로 성능을 극대화하는 실전 노하우를 단계별로 알려드립니다.

데이터 파이프라인을 구축하고 운영하는 실무자라면 누구나 한 번쯤 Airflow DAG의 느린 실행 시간 때문에 애를 먹었던 경험이 있을 것입니다. 특정 DAG가 예상보다 오래 걸리거나, 리소스 문제로 실패하기라도 하면 전체 데이터 흐름에 병목이 생기고, 이는 곧 서비스 지연이나 데이터 누락으로 이어질 수 있습니다.

특히 실무 1~3년차 주니어 개발자에게는 이러한 성능 문제를 해결하는 것이 막막하게 느껴질 수 있습니다. 기본기는 갖췄지만, 실제 운영 환경에서 DAG 성능 최적화를 어떻게 시작해야 할지, 어떤 기법들을 적용해야 효과적인지 감을 잡기 어렵기 때문입니다.

이 글에서는 Airflow DAG의 실행 시간을 단축하기 위한 실용적인 태스크 의존성 최적화 및 병렬 처리 기법들을 단계별로 소개합니다. 단순히 이론을 나열하는 것을 넘어, 실제 프로젝트에서 겪었던 시행착오와 해결 경험을 바탕으로, 여러분이 직면할 수 있는 문제들을 효과적으로 해결할 수 있도록 돕겠습니다.

📑 목차

• 1단계: Airflow DAG 성능 저하의 원인 이해하기
• 1. 불필요하거나 과도한 태스크 의존성
• 2. 비효율적인 태스크 설계
• 3. 제한적인 병렬 처리 및 리소스 부족
• 2단계: 기본기 다지기 - 태스크 의존성 재정의 및 간소화
• 1. 불필요한 의존성 제거
• 2. 태스크 병합 및 분리 전략
• 3단계: 본격적인 속도전 - 병렬 처리 기법 도입
• 1. TaskGroup을 활용한 논리적 그룹화 및 병렬 실행
• 2. 동적 태스크 매핑 (Dynamic Task Mapping) 활용
• 4단계: 고급 기법 - Executor 선택 및 리소스 관리
• 1. 적절한 Executor 선택
• 2. 리소스 풀 (Pool) 활용
• 3. 트리거 룰 (trigger_rule) 활용
• 5단계: 최적화의 완성 - 모니터링 및 지속적인 개선
• 1. Airflow UI를 통한 모니터링
• 2. 메트릭 수집 및 시각화
• 3. A/B 테스트 및 벤치마킹
• 마무리하며: Airflow DAG 최적화, 지속적인 관심이 핵심

Image by AlexKantsur on Pixabay

1단계: Airflow DAG 성능 저하의 원인 이해하기

본격적인 최적화에 앞서, 우리 DAG가 왜 느려지는지 그 근본적인 원인을 이해하는 것이 중요합니다. 대부분의 경우, 성능 저하는 다음 세 가지 주요 원인 중 하나 이상에서 비롯됩니다.

1. 불필요하거나 과도한 태스크 의존성

Airflow는 태스크 간의 의존성을 기반으로 작업을 스케줄링합니다. 하지만 때로는 너무 많은 태스크가 서로를 기다리게 만들거나, 실제로는 독립적으로 실행될 수 있는 태스크들이 불필요하게 직렬로 연결되어 있는 경우가 많습니다. 예를 들어, 데이터 적재와 관련된 여러 태스크가 있는데, 이들이 모두 개별적으로 완료되어야 다음 단계로 넘어갈 수 있도록 설정되어 있다면, 앞선 태스크 중 하나라도 지연될 경우 전체 DAG의 실행 시간이 선형적으로 증가합니다. 잘못된 의존성 설정은 DAG의 병목 지점을 만들고, 리소스 활용 효율성을 떨어뜨리는 주범입니다.

2. 비효율적인 태스크 설계

각 태스크 자체가 비효율적으로 설계되었을 수도 있습니다. 예를 들어, 너무 많은 데이터를 한 번에 처리하려 하거나, 데이터베이스 쿼리가 최적화되지 않았거나, 외부 API 호출에 과도한 지연이 발생하는 등의 문제들이 있습니다. 또한, 작은 작업을 여러 태스크로 쪼개는 것은 좋지만, 너무 잘게 쪼개어 태스크 간의 오버헤드(컨텍스트 스위칭, 리소스 할당 등)가 커지는 경우도 발생합니다. 이 경우, 태스크 자체의 실행 시간은 짧더라도 총 오버헤드가 DAG 실행 시간을 늘릴 수 있습니다.

3. 제한적인 병렬 처리 및 리소스 부족

Airflow는 기본적으로 태스크 병렬 처리를 지원하지만, 이를 제대로 활용하지 못하는 경우가 많습니다. Airflow 인스턴스의 Executor 설정(예: `max_active_tasks_per_dag`, `parallelism`), 워커 노드의 리소스(CPU, 메모리) 부족, 그리고 DAG 내에서 병렬 처리를 명시적으로 정의하지 않음으로써 발생하는 문제가 대표적입니다. 병렬 처리를 제대로 활용하지 못하면, 아무리 독립적인 태스크라도 직렬로 실행될 수밖에 없어 실행 시간이 길어집니다.

2단계: 기본기 다지기 - 태스크 의존성 재정의 및 간소화

가장 먼저 시도해야 할 것은 현재 DAG의 태스크 의존성을 면밀히 검토하고 불필요한 직렬화 부분을 제거하는 것입니다.

1. 불필요한 의존성 제거

현재 DAG의 그래프 뷰를 통해 태스크 흐름을 시각적으로 확인해 보세요. 특정 태스크가 다른 태스크의 결과에 의존하지 않는데도 불구하고 순서대로 실행되도록 설정되어 있지는 않나요? 예를 들어, '상품 데이터 수집' 태스크와 '주문 데이터 수집' 태스크는 서로 독립적으로 실행될 수 있음에도 불구하고, 한쪽이 끝나야 다른 쪽이 시작되도록 설정된 경우가 많습니다.

이러한 경우, 의존성을 끊고 두 태스크가 동시에 실행될 수 있도록 변경하면 전체 DAG 실행 시간을 크게 단축할 수 있습니다.

# 기존 코드 (직렬 실행)
# collect_product_data >> collect_order_data >> process_data

# 변경 코드 (병렬 실행)
# 두 태스크를 동시에 시작
[collect_product_data, collect_order_data] >> process_data

위 예시처럼 리스트를 활용하면 여러 태스크를 한 번에 다음 단계로 연결하거나, 동시에 실행시킬 수 있습니다.

2. 태스크 병합 및 분리 전략

때로는 너무 작은 태스크들이 연쇄적으로 연결되어 있어, 각 태스크 실행 시 발생하는 오버헤드(태스크 상태 업데이트, 리소스 할당 등)가 실제 작업 시간보다 길어지는 경우가 있습니다. 이럴 때는 여러 개의 작은 태스크를 하나의 더 큰 태스크로 병합하는 것을 고려해볼 수 있습니다. 예를 들어, 여러 개의 작은 SQL 쿼리를 하나의 배치 쿼리로 묶거나, 파이썬 스크립트 내에서 여러 단계를 처리하도록 하는 방식입니다.

반대로, 하나의 태스크가 너무 많은 작업을 처리하여 병목이 되는 경우도 있습니다. 이런 태스크는 여러 개의 독립적인 서브 태스크로 분리하여 병렬로 처리할 수 있는 여지를 만들 수 있습니다. 중요한 것은 태스크의 '적절한 크기'를 찾는 것입니다. 이는 데이터 처리량, 로직의 복잡성, 태스크 간의 의존성 등을 종합적으로 고려하여 결정해야 합니다.

3단계: 본격적인 속도전 - 병렬 처리 기법 도입

태스크 의존성을 최적화했다면, 이제 Airflow의 강력한 병렬 처리 기능을 활용할 차례입니다.

1. TaskGroup을 활용한 논리적 그룹화 및 병렬 실행

Airflow 2.0부터 도입된 TaskGroup은 여러 태스크를 논리적으로 묶어 DAG UI를 깔끔하게 정리해줄 뿐만 아니라, 그룹 내 태스크들을 병렬로 실행할 수 있는 강력한 도구입니다. 이전에는 `SubDagOperator`를 사용했지만, `SubDagOperator`는 여러 가지 오버헤드와 제약사항(별도의 스케줄링, 독립적인 실행 등)이 많아 TaskGroup이 대부분의 경우 훨씬 선호됩니다.

TaskGroup을 사용하면 특정 기능 단위로 태스크들을 묶고, 이 그룹 내의 태스크들은 병렬로 실행되도록 구성할 수 있습니다.

from airflow.models.dag import DAG
from airflow.operators.bash import BashOperator
from airflow.utils.task_group import TaskGroup
from datetime import datetime

with DAG(
    dag_id='taskgroup_parallel_example',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=['example', 'optimization']
) as dag:
    start_task = BashOperator(task_id='start', bash_command='echo "Starting processing"')

    with TaskGroup("data_ingestion_group") as ingestion_group:
        ingest_users = BashOperator(task_id='ingest_users', bash_command='sleep 5 && echo "Ingesting users"')
        ingest_products = BashOperator(task_id='ingest_products', bash_command='sleep 7 && echo "Ingesting products"')
        ingest_orders = BashOperator(task_id='ingest_orders', bash_command='sleep 3 && echo "Ingesting orders"')
        # 이 세 태스크는 병렬로 실행될 수 있습니다.

    process_data = BashOperator(task_id='process_all_data', bash_command='echo "Processing all ingested data"')

    start_task >> ingestion_group >> process_data

위 예시에서 `ingestion_group` 내의 세 태스크는 동시에 실행됩니다. 만약 이들이 직렬로 실행되었다면 최소 5+7+3=15초가 걸렸겠지만, 병렬로 실행되면 가장 오래 걸리는 태스크인 `ingest_products`의 실행 시간인 7초 내외로 그룹이 완료됩니다.

SubDagOperator와 TaskGroup의 주요 차이점은 다음과 같습니다.

특징	SubDagOperator	TaskGroup
Airflow UI 표시	별도의 DAG 그래프로 표시	메인 DAG 내에 중첩되어 표시
스케줄링	자체적으로 스케줄러에 등록/실행	메인 DAG의 일부로 스케줄링
오버헤드	높음 (별도의 DAG Run 생성)	낮음 (단일 DAG Run 내에서 처리)
권장 사용처	구조적으로 완전히 독립된 서브 프로세스 (거의 사용 안 함)	논리적 그룹화 및 병렬 처리가 필요한 경우 (대부분의 경우)

2. 동적 태스크 매핑 (Dynamic Task Mapping) 활용

처리해야 할 아이템의 개수가 실행 시점에 결정되는 경우, 예를 들어 특정 디렉토리 내의 모든 파일을 처리해야 하거나, 특정 큐의 메시지들을 처리해야 하는 경우에 동적 태스크 매핑 (Dynamic Task Mapping)은 매우 강력한 도구입니다. Airflow 2.3부터 정식 도입된 이 기능은 상위 태스크의 출력(예: 파일 목록)을 하위 태스크의 입력으로 받아, 해당 목록의 각 아이템에 대해 동적으로 여러 개의 태스크 인스턴스를 생성하고 병렬로 실행할 수 있게 해줍니다.

이는 `for` 루프를 사용하여 수동으로 태스크를 생성하는 것보다 훨씬 효율적이고, DAG 코드를 간결하게 유지할 수 있게 합니다.

from airflow.models.dag import DAG
from airflow.decorators import task
from datetime import datetime

with DAG(
    dag_id='dynamic_mapping_example',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=['example', 'optimization']
) as dag:
    @task
    def get_file_list():
        # 실제 환경에서는 S3, GCS 등에서 파일 목록을 가져올 것입니다.
        return ["file_A.csv", "file_B.csv", "file_C.csv", "file_D.csv"]

    @task
    def process_file(filename: str):
        # 각 파일을 처리하는 로직
        print(f"Processing {filename}...")
        import time
        time.sleep(2) # 파일 처리 시간 시뮬레이션
        return f"Processed {filename}"

    files = get_file_list()
    processed_results = process_file.partial().expand(filename=files) # 여기서 동적 매핑 발생

    # processed_results는 각각의 process_file 태스크 인스턴스의 결과 목록을 담게 됩니다.

`process_file.partial().expand(filename=files)` 부분이 핵심입니다. `get_file_list` 태스크가 반환하는 파일 목록의 각 요소에 대해 `process_file` 태스크가 병렬로 실행됩니다. 이는 처리해야 할 데이터의 양이 가변적일 때 최적의 병렬 처리를 가능하게 합니다.

Image by chief39 on Pixabay

4단계: 고급 기법 - Executor 선택 및 리소스 관리

Airflow의 성능은 어떤 Executor를 사용하고, 리소스를 어떻게 관리하느냐에 따라 크게 달라질 수 있습니다.

1. 적절한 Executor 선택

Airflow는 다양한 Executor를 지원하며, 각 Executor는 태스크를 실행하는 방식과 병렬 처리 능력에서 차이가 있습니다.

Executor	특징	장점	단점	권장 환경
SequentialExecutor	단일 프로세스에서 태스크 순차 실행	가장 간단, 디버깅 용이	병렬 처리 불가, 성능 매우 낮음	로컬 개발/테스트 환경
LocalExecutor	단일 머신에서 멀티 프로세스로 태스크 병렬 실행	설정 간단, 효율적인 리소스 활용	단일 장애점, 확장성 제한	소규모 프로덕션, 온프레미스
CeleryExecutor	분산 큐 (Celery) 기반으로 태스크 분산 실행	높은 확장성, 워커 추가 용이	메시지 큐(Redis/RabbitMQ) 필요, 설정 복잡	중대규모 프로덕션, 동적인 워크로드
KubernetesExecutor	각 태스크를 K8s Pod로 실행	태스크별 격리, 강력한 리소스 관리, 클라우드 네이티브	Kubernetes 클러스터 필요, 설정 복잡	대규모 프로덕션, 클라우드 환경

대부분의 프로덕션 환경에서는 CeleryExecutor나 KubernetesExecutor를 사용하여 높은 확장성과 안정성을 확보합니다. 주니어 개발자라면 현재 운영 환경의 Executor가 무엇인지 확인하고, 해당 Executor의 최대 병렬 태스크 수(`max_active_tasks_per_dag`, `parallelism` 설정)를 적절히 조절하여 리소스 제약을 넘어서지 않도록 관리하는 것이 중요합니다.

2. 리소스 풀 (Pool) 활용

특정 유형의 태스크(예: 특정 DB에 과부하를 줄 수 있는 쿼리, 제한된 API 호출)가 시스템에 과도한 부하를 주지 않도록 관리해야 할 때 Airflow Pool을 사용할 수 있습니다. Pool은 동시에 실행될 수 있는 태스크의 수를 제한하는 메커니즘입니다.

Airflow UI에서 Pool을 생성하고, DAG 내의 태스크에 `pool='my_resource_pool'` 파라미터를 지정하면 됩니다. 예를 들어, 동시에 5개 이상의 DB 쿼리가 실행되지 않도록 제한할 수 있습니다.

# Airflow UI -> Admin -> Pools 에서 'db_query_pool' 생성, 슬롯 5개 할당
db_query_task_1 = BashOperator(task_id='db_query_1', bash_command='sleep 10', pool='db_query_pool')
db_query_task_2 = BashOperator(task_id='db_query_2', bash_command='sleep 10', pool='db_query_pool')
db_query_task_3 = BashOperator(task_id='db_query_3', bash_command='sleep 10', pool='db_query_pool')
db_query_task_4 = BashOperator(task_id='db_query_4', bash_command='sleep 10', pool='db_query_pool')
db_query_task_5 = BashOperator(task_id='db_query_5', bash_command='sleep 10', pool='db_query_pool')
db_query_task_6 = BashOperator(task_id='db_query_6', bash_command='sleep 10', pool='db_query_pool')

# 이 태스크들은 db_query_pool에 할당된 5개의 슬롯 내에서만 병렬 실행됩니다.
# 6번째 태스크는 앞선 태스크 중 하나가 완료되어 슬롯이 비워질 때까지 대기합니다.

이는 성능 최적화와 동시에 시스템 안정성을 확보하는 중요한 기법입니다.

3. 트리거 룰 (trigger_rule) 활용

각 태스크에는 trigger_rule이라는 파라미터가 있습니다. 기본값은 `all_success`로, 모든 상위 태스크가 성공해야 해당 태스크가 실행됩니다. 하지만 `all_done`, `one_success`, `none_failed_min_one_success` 등 다양한 룰을 활용하면 유연한 DAG 흐름을 만들고, 불필요한 대기 시간을 줄일 수 있습니다.

• `all_success` (기본값): 모든 상위 태스크가 성공해야 실행됩니다.
• `all_done`: 모든 상위 태스크가 어떤 상태(성공, 실패, 스킵 등)로든 완료되면 실행됩니다. 일부 태스크가 실패해도 다음 단계로 넘어갈 수 있습니다.
• `one_success`: 하나 이상의 상위 태스크가 성공하면 실행됩니다. 여러 경로 중 하나만 성공해도 되는 경우 유용합니다.

예를 들어, A 또는 B 작업 중 하나만 성공해도 다음 C 작업으로 넘어갈 수 있다면 `trigger_rule='one_success'`를 C 작업에 적용하여 전체 DAG의 완료 시간을 단축할 수 있습니다.

Image by cocoandwifi on Pixabay

5단계: 최적화의 완성 - 모니터링 및 지속적인 개선

최적화는 한 번의 작업으로 끝나지 않습니다. 변경 사항을 적용한 후에는 반드시 그 효과를 측정하고, 지속적으로 개선해나가야 합니다.

1. Airflow UI를 통한 모니터링

Airflow UI의 Graph View, Gantt Chart, Task Duration 등을 통해 각 태스크의 실행 시간과 전체 DAG의 흐름을 시각적으로 파악할 수 있습니다. Gantt Chart는 태스크들이 언제 시작하고 끝나는지, 얼마나 겹쳐서 실행되는지 등을 한눈에 보여주므로 병목 지점을 찾고 병렬 처리 효과를 확인하는 데 매우 유용합니다.

• Graph View: 태스크 간 의존성 및 흐름 확인
• Gantt Chart: 태스크 실행 시간 및 병렬 처리 여부 시각화
• Task Duration: 각 태스크의 평균 실행 시간 변화 추이 확인

특히 Task Duration을 주기적으로 확인하여 특정 태스크의 실행 시간이 갑자기 늘어나지는 않았는지, 최적화 작업 후 실제로 줄어들었는지를 검토해야 합니다.

2. 메트릭 수집 및 시각화

더 정교한 모니터링을 위해서는 Prometheus, Grafana와 같은 외부 모니터링 도구를 활용하는 것이 좋습니다. Airflow는 자체적으로 많은 메트릭을 Prometheus 형식으로 노출할 수 있으며, 이를 Grafana 대시보드에서 시각화하여 DAG 실행 시간, 태스크 성공/실패율, 워커 리소스 사용량 등을 실시간으로 확인할 수 있습니다.

주요 모니터링 지표:

• DAG Run Duration: 전체 DAG 실행 시간
• Task Instance Duration: 개별 태스크 실행 시간
• Scheduler Heartbeat: 스케줄러의 건강 상태
• Worker CPU/Memory Usage: 워커 리소스 사용량
• Queue Length: Executor 큐에 대기 중인 태스크 수

3. A/B 테스트 및 벤치마킹

중요한 DAG에 대한 최적화 작업을 수행할 때는, 기존 DAG와 최적화된 DAG를 동시에 운영하여 A/B 테스트를 진행하는 것을 고려해볼 수 있습니다. 이를 통해 실제 환경에서의 성능 개선 효과를 정확하게 측정하고, 문제가 발생했을 때 빠르게 롤백할 수 있는 안전장치를 마련할 수 있습니다. 벤치마킹 도구를 사용하여 다양한 부하 조건에서 DAG의 성능을 측정하는 것도 좋습니다.

마무리하며: Airflow DAG 최적화, 지속적인 관심이 핵심

Airflow DAG의 실행 시간 단축은 단순히 '빨리 끝내는 것'을 넘어, 리소스 효율성 증대, 데이터 파이프라인의 안정성 확보, 그리고 궁극적으로는 운영 비용 절감과 직결되는 중요한 작업입니다.

실무 1~3년차 주니어 개발자로서, 처음부터 완벽한 최적화를 목표로 하기보다는, 현재 운영 중인 DAG의 병목을 찾아내고, 이 글에서 소개한 태스크 의존성 최적화와 병렬 처리 기법들을 하나씩 적용해보는 것부터 시작해보세요. Airflow UI의 Gantt Chart를 활용하여 변경 전후의 실행 시간을 비교하고, 성공적인 개선 경험을 쌓아나가는 것이 중요합니다.

데이터 엔지니어링 환경은 끊임없이 변화하므로, DAG 최적화는 지속적인 관심과 개선이 필요한 여정입니다. 이 글이 여러분의 Airflow DAG 성능 튜닝 여정에 실질적인 도움이 되었기를 바랍니다. 혹시 이 외에 효과적인 Airflow 최적화 팁이 있다면 댓글로 공유해 주세요!

📌 함께 읽으면 좋은 글

• [생산성 자동화] Git Hooks로 개발 워크플로우 자동화: 코드 품질과 커밋 메시지 검증 완벽 가이드
• [AI 머신러닝] LLM 에이전트 개발 실전 가이드: 자율적인 태스크 수행 시스템 구축 전략
• [AI 머신러닝] RAG 시스템 구축을 위한 벡터 데이터베이스 활용 전략: LLM 성능 향상 실전 가이드

이 글이 도움이 되셨다면 공감(♥)과 댓글로 응원해 주세요!
궁금한 점이나 다루었으면 하는 주제가 있다면 댓글로 남겨주세요.