[ Airflow ] task_concurrency - 동일 작업 개수 제한하기

task_concurrency는 Airflow에서 개별 Task가 동시에 실행될 수 있는 최대 인스턴스 수를 제한하는 파라미터입니다. 이를 통해, 특정 작업에 대해 병렬 실행 수를 제어하고, 리소스를 효율적으로 사용할 수 있습니다.

task_concurrency는 주로 리소스 소모가 많은 작업이나 외부 시스템과의 통신을 포함하는 작업에서 유용합니다. 예를 들어, 외부 API에 요청을 보내거나, 파일 시스템에서 데이터를 읽고 쓸 때, 이 값은 동시에 실행되는 작업 수를 제한하여 시스템 과부하를 방지할 수 있습니다.

1. task_concurrency 사용법

task_concurrency는 개별 Task에 대해 설정할 수 있으며, 해당 Task가 동시에 실행될 수 있는 최대 수를 정의합니다. DAG 내의 다른 Task들은 task_concurrency의 영향을 받지 않으므로, 각 Task별로 독립적으로 설정할 수 있습니다.

예제: task_concurrency 설정을 사용한 DAG

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from datetime import datetime, timedelta

# DAG 정의
dag = DAG(
    'my_dag_with_task_concurrency',
    start_date=datetime(2023, 1, 1),
    schedule_interval=timedelta(days=1),  # 하루에 한 번 실행
)

# task1은 최대 2개 인스턴스가 동시에 실행될 수 있음
task1 = DummyOperator(
    task_id='task1',
    dag=dag,
    task_concurrency=2,  # 동시에 실행되는 최대 task1 수를 2로 제한
)

# task2는 최대 1개 인스턴스만 실행될 수 있음
task2 = DummyOperator(
    task_id='task2',
    dag=dag,
    task_concurrency=1,  # 동시에 실행되는 최대 task2 수를 1로 제한
)

# task3는 제한 없이 실행될 수 있음
task3 = DummyOperator(
    task_id='task3',
    dag=dag,
)

# task 의존성 설정
task1 >> task2 >> task3

예제 설명

• task_concurrency=2: task1은 동시에 최대 2개 인스턴스만 실행될 수 있습니다. 즉, task1이 두 개 이상 동시에 실행되면, 새로운 인스턴스는 대기 상태에 놓이게 됩니다.
• task_concurrency=1: task2는 동시에 최대 1개 인스턴스만 실행됩니다. 즉, task2는 하나의 인스턴스만 실행되며, 다른 인스턴스가 실행될 때까지 대기해야 합니다.
• task3에는 task_concurrency가 설정되지 않았으므로, 기본값인 제한 없음 상태로 실행됩니다.

이 설정은, 특정 작업이 리소스를 과도하게 소비하거나 외부 시스템(예: 데이터베이스, API, 파일 시스템 등)과 통신할 때 유용합니다. 예를 들어, 외부 API의 호출 수에 제한이 있을 때, 이 값을 설정하여 API의 호출 수가 과도해지지 않도록 제어할 수 있습니다.

2. task_concurrency 활용 시나리오

1. 외부 API 호출 제한: 외부 API의 호출 제한이 있을 때, task_concurrency를 사용하여 한 번에 너무 많은 API 호출이 이루어지지 않도록 제한할 수 있습니다.
2. 파일 시스템 리소스 관리: 파일 시스템이나 데이터베이스에 액세스할 때, 동시에 여러 작업이 실행되면 리소스가 과부하 될 수 있습니다. task_concurrency를 사용하여 동시에 실행되는 작업 수를 제한하면 이러한 문제를 피할 수 있습니다.
3. 한정된 리소스 사용: 예를 들어, 하나의 시스템에서 여러 작업이 동시에 실행되면 성능 저하가 발생할 수 있습니다. 이 경우, 특정 Task에 대해 동시 실행 수를 제한하면 시스템 자원을 효율적으로 사용할 수 있습니다.

3. task_concurrency의 우선순위

task_concurrency는 다음과 같은 우선순위를 가집니다:

1. task_concurrency가 설정된 경우, 해당 Task는 최대 설정된 수의 작업만 실행할 수 있습니다.
2. DAG 수준의 concurrency가 설정되어 있을 때, concurrency는 DAG 내 모든 Task의 동시 실행 수를 제한합니다.
3. max_active_runs가 설정된 경우, 여러 DAG 인스턴스가 동시에 실행되는 수를 제한하지만, 개별 Task에서의 실행 수를 제한하지 않습니다.

따라서, 여러 제약이 있을 때 task_concurrency가 우선적으로 적용되고, 그다음에 concurrency와 max_active_runs가 함께 동작합니다.

 

4. 결론

• task_concurrency는 개별 Task에서 동시에 실행될 수 있는 작업 수를 제한하는 파라미터입니다.
• 주로 외부 시스템과의 통신이나 리소스가 한정된 작업에서 사용되며, 시스템의 과부하를 방지하는 데 유용합니다.
• task_concurrency를 적절하게 설정하면, 여러 작업이 동시에 실행될 때 발생할 수 있는 리소스 경쟁을 줄일 수 있습니다.

따라서, 리소스를 효율적으로 사용할 수 있도록 각 Task의 특성에 맞게 task_concurrency를 설정하는 것이 중요합니다.