Airflow 태스크 100개를 걸었는데 왜 일부만 돌까?

Q: Airflow 태스크 100개를 걸었는데 왜 일부만 돌까?

parallelism 8, max_active_tasks 4에서 태스크 100개짜리 run의 running은 세 번 모두 4였으니, 동시 실행 수는 여러 한도 중 가장 작은 값에서 멈춘다.

이번 시간에는 서로 의존성이 없는 태스크 100개짜리 DAG를 Docker로 띄운 Airflow에서 돌려 보고, 동시 실행 수를 막는 한도가 어디서 걸리는지 메타데이터 DB를 1초마다 세어 가며 배워가는 시간입니다.

예를 들어서 우리가 태스크 100개짜리 DAG를 만들어 실행했다고 해볼게요. 100개가 한꺼번에 돌 거라고 기대했는데, running인 태스크는 몇 개뿐이고 나머지는 실행을 기다려요. 실패한 태스크도 없고 오류 로그도 없습니다.

Airflow에는 동시에 실행할 태스크 수를 막는 설정이 여러 곳에 있고, 범위가 서로 다르기 때문이에요. 동시에 도는 태스크 수는 그중 가장 작은 값에서 멈춥니다.

한도는 다섯 군데에 있습니다

parallelism ([core] parallelism): scheduler 하나가 동시에 실행하는 태스크 수의 상한이에요. 모든 DAG의 태스크를 합친 수입니다.
max_active_runs (DAG 인자, 기본값은 [core] max_active_runs_per_dag): DAG 하나에서 동시에 running일 수 있는 run 수예요.
max_active_tasks (DAG 인자, 기본값은 [core] max_active_tasks_per_dag): run 하나 안에서 동시에 running이거나 queued일 수 있는 태스크 수예요.
max_active_tis_per_dag (태스크 인자, 기본값 없음): 태스크 하나가 여러 run에 걸쳐 동시에 실행되는 수예요.
pool: 이름을 붙인 slot 묶음이에요. 태스크는 pool 하나에 속하고, 실행되는 동안 그 pool의 slot을 pool_slots(기본 1)만큼 차지합니다. pool을 지정하지 않은 태스크는 default_pool에 들어가요.

3.3.2의 기본값은 parallelism 32, max_active_tasks_per_dag 16, max_active_runs_per_dag 16이에요. 그러니 기본 설정에서 태스크 100개짜리 DAG의 run 하나는 max_active_tasks 16에 먼저 걸려요. 32인 parallelism까지 가지 않습니다.

실습 준비

이 글의 명령은 모두 실습 코드의 airflow-concurrency-limits/ 디렉터리에서 실행합니다.

compose.yaml: PostgreSQL 메타데이터 DB와 Airflow standalone 서비스 정의
dags/: 실습 DAG 파일 2개(wide_100.py, pools.py)와 공통 값 common.py(파일 3개)

컨테이너는 두 개예요. PostgreSQL은 Airflow 메타데이터 DB로만 쓰고, Airflow는 airflow standalone 명령으로 scheduler, API 서버, DAG 프로세서, triggerer를 한 컨테이너에서 실행해요. standalone은 개발용 실행 방식이에요.

compose.yaml은 postgres(postgres:17, 호스트 포트 7631)와 airflow(apache/airflow:3.3.2-python3.12, 호스트 포트 7630) 두 서비스를 정의합니다.

AIRFLOW__CORE__EXECUTOR: 태스크를 Airflow 컨테이너 안의 프로세스로 실행하는 LocalExecutor를 써요.
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION: 새 DAG를 일시 정지 상태로 등록하지 않게 해요. 그래서 unpause 없이 바로 trigger할 수 있어요.
AIRFLOW__CORE__PARALLELISM, AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG: 실습마다 바꾸는 두 한도예요. 셸 변수 PAR, MAT로 값을 넣고, 주지 않으면 8과 4를 씁니다.

기본값 대신 8과 4를 쓰는 건, LocalExecutor가 워커를 parallelism 수만큼 띄우고 태스크마다 프로세스가 하나 더 생겨서 기본값으로는 이 컨테이너 메모리 상한에 여유가 거의 없기 때문이에요. "parallelism이 max_active_tasks보다 크다"는 기본값의 관계는 그대로 둡니다.

DAG의 태스크는 모두 sleep만 하는 BashOperator이고, 태스크 하나가 sleep하는 시간 SLEEP은 60초입니다. dags/wide_100.py는 서로 의존성이 없는 태스크 100개를 만들어요. 한도를 코드에서 정하지 않기 때문에, 이 DAG의 max_active_tasks는 [core] max_active_tasks_per_dag 값을 따라가요. 모든 DAG는 schedule=None이라 직접 trigger할 때만 run이 생깁니다.

이제 명령을 짧게 쓸 수 있는 셸 함수를 만들어 놓을게요. COMPOSE_PROJECT_NAME으로 compose 프로젝트 이름을 afcon으로 고정해요. 아래 블록은 파일로 만들지 말고 실습 디렉터리에서 연 터미널에 그대로 붙여 넣어 주세요. 이후 명령은 모두 이 터미널에서 실행하고, 터미널을 새로 열면 다시 붙여 넣으시면 됩니다.

export COMPOSE_PROJECT_NAME=afcon

af() { docker compose exec -T airflow airflow "$@"; }

meta() { docker compose exec -T postgres psql -U postgres -d airflow "$@"; }

waitup() { until curl -sf localhost:7630/api/v2/monitor/health > /dev/null; do sleep 3; done; }

limits() {
  PAR=$1 MAT=$2 docker compose up -d airflow 2> /dev/null
  waitup
  echo "parallelism=$(af config get-value core parallelism)" \
       "max_active_tasks_per_dag=$(af config get-value core max_active_tasks_per_dag)"
}

watch_states() {
  local ids; ids=$(printf "'%s'," "$@"); ids=${ids%,}
  echo "time     running queued scheduled success | run_running run_queued"
  meta -tA -F ' ' <<EOF | awk 'NF { printf "%s %7s %6s %9s %7s | %11s %10s\n", $1, $2, $3, $4, $5, $6, $7; fflush() }'
SELECT to_char(clock_timestamp(), 'HH24:MI:SS'),
       count(*) FILTER (WHERE state = 'running'),
       count(*) FILTER (WHERE state = 'queued'),
       count(*) FILTER (WHERE state = 'scheduled'),
       count(*) FILTER (WHERE state = 'success'),
       (SELECT count(*) FROM dag_run WHERE dag_id IN ($ids) AND state = 'running'),
       (SELECT count(*) FROM dag_run WHERE dag_id IN ($ids) AND state = 'queued')
FROM task_instance WHERE dag_id IN ($ids)
HAVING count(*) FILTER (WHERE state IS NULL OR state IN ('scheduled', 'queued', 'running')) > 0
    OR (SELECT count(*) FROM dag_run WHERE dag_id IN ($ids) AND state IN ('queued', 'running')) > 0
\watch interval=1 min_rows=1 ${SECS:+count=$SECS}
EOF
}

maxrun() { awk 'NR > 1 && $2 > m { m = $2 } END { print "max running =", m }' "$1"; }

pool_usage() {
  local busy="SELECT count(*) FROM task_instance WHERE pool = '$1' AND (state IS NULL OR state IN ('scheduled', 'queued', 'running'))"
  until [ "$(meta -tAc "$busy")" -gt 0 ]; do sleep 1; done
  echo "time     dag_id     running queued scheduled slots_in_use"
  meta -tA -F ' ' <<EOF | awk 'NF { printf "%s %-10s %7s %6s %9s %12s\n", $1, $2, $3, $4, $5, $6; fflush() }'
SELECT to_char(clock_timestamp(), 'HH24:MI:SS'), dag_id,
       count(*) FILTER (WHERE state = 'running'),
       count(*) FILTER (WHERE state = 'queued'),
       count(*) FILTER (WHERE state = 'scheduled'),
       (SELECT coalesce(sum(pool_slots), 0) FROM task_instance
         WHERE pool = '$1' AND state IN ('queued', 'running'))
FROM task_instance
WHERE pool = '$1' AND (state IS NULL OR state IN ('scheduled', 'queued', 'running'))
GROUP BY dag_id ORDER BY dag_id
\watch interval=1 min_rows=1
EOF
}

procs() {
  docker top afcon-airflow-1 -eo pid,rss,args \
    | awk 'NR > 1 { if ($0 ~ /LocalExecutor/) w++; if ($0 ~ /sleep [0-9]/) s++ }
           END { print "workers =", w + 0, " sleeping tasks =", s + 0 }'
}

mem() { docker stats --no-stream --format '{{.Name}} {{.MemUsage}} {{.CPUPerc}}' | grep afcon; }

limits <parallelism> <max_active_tasks_per_dag>: 두 한도를 셸 변수로 넘겨 Airflow 컨테이너만 다시 만들고, 적용된 값을 config get-value로 출력해요.
watch_states <dag_id>...: 1초마다 task_instance 테이블에서 태스크를 상태별로 세고, dag_run 테이블에서 running과 queued인 run을 세요. 끝나지 않은 태스크와 run이 없어지면 멈춥니다. SECS를 주면 그 횟수만큼만 봐요(SECS=100 watch_states ...이면 100초).
maxrun: watch_states 출력에서 running의 최댓값을 뽑아요.
pool_usage <pool>: pool 하나를 쓰는 태스크를 DAG별로 세고, running과 queued인 태스크가 차지한 slot 합을 함께 봐요.
procs: Airflow 컨테이너의 프로세스 목록에서 LocalExecutor 워커 프로세스와 실행 중인 sleep 명령을 세요.
mem: 이 실습 컨테이너들의 메모리와 CPU 사용량이에요.

상태를 1초 간격으로 세기 때문에, 1초보다 짧게 지나가는 상태는 놓칠 수 있어요. 출력의 시간은 모두 UTC예요.

실습 1: 태스크 100개를 돌리면

parallelism=8, max_active_tasks_per_dag=4로 wide_100을 trigger하고 끝날 때까지 상태를 기록해요. 한 번에 25분쯤 걸려요. trigger부터 maxrun까지를 파일 이름의 번호만 바꿔 세 번 실행했습니다.

docker compose up -d 2> /dev/null
limits 8 4
mkdir -p out
af dags trigger wide_100 > /dev/null
watch_states wide_100 > out/1-states-1.txt
maxrun out/1-states-1.txt

parallelism=8 max_active_tasks_per_dag=4
max running = 4

running은 4에서 멈췄고, 2회차와 3회차도 4였어요. parallelism은 8이니, run 하나의 한도인 max_active_tasks 4에 걸린 거예요. 기록의 처음과 첫 4개가 끝나는 무렵은 이렇습니다.

time     running queued scheduled success | run_running run_queued
14:53:18       0      4        96       0 |           1          0
14:53:19       4      0        96       0 |           1          0
14:53:20       4      0        96       0 |           1          0

14:54:18       4      0        96       0 |           1          0
14:54:19       0      0        96       4 |           1          0
14:54:20       0      4        92       4 |           1          0
14:54:21       4      0        92       4 |           1          0

기다리는 태스크 96개는 scheduled예요. queued는 4개가 running이 되기 직전에 1초쯤만 보여요.

scheduled: scheduler가 의존성이 모두 충족돼 실행해야 한다고 판단한 상태
queued: executor에 넘겨져 워커를 기다리는 상태

3.3.2 scheduler는 scheduled 태스크 가운데 한도에 걸리지 않은 것만 골라 queued로 바꿔요. 그래서 max_active_tasks에 걸린 run의 태스크는 queued로 바뀌지 않고 scheduled로 남습니다.

scheduler 로그에는 max_active_tasks에 걸렸다는 문장이 있어요. 3.3.2 소스에서 INFO로 찍는 문장입니다.

docker compose logs airflow | grep -c "max_active_tasks limit"

그런데 세 번, 75번의 교대 동안 4줄뿐이에요. 3.3.2 scheduler는 태스크를 고르는 SQL에서, 이미 running과 queued 합이 max_active_tasks에 이른 run을 빼고 조회해요. 조회 결과에 없는 태스크에 대해서는 이 문장을 찍지 않으니, 로그가 없어도 max_active_tasks에 걸려 있을 수 있어요. 그러니 running 수를 한도와 직접 비교해 주셔야 합니다.

실습 2: parallelism을 올려도 그대로입니다

이번에는 한도를 바꿔 가며 wide_100을 100초씩 볼게요. 앞 run이 섞이지 않게 매번 DB까지 새로 만들어요. max_active_tasks에 걸렸다는 로그 줄 수도 함께 셉니다.

try_limits() {
  echo; echo "===== parallelism=$1 max_active_tasks_per_dag=$2"
  docker compose down -v 2> /dev/null
  PAR=$1 MAT=$2 docker compose up -d 2> /dev/null
  limits $1 $2
  sleep 30
  echo "-- idle"; mem | grep airflow; procs
  af dags trigger wide_100 > /dev/null
  SECS=100 watch_states wide_100 > "out/2-states-$1-$2.txt"
  maxrun "out/2-states-$1-$2.txt"
  echo "-- running"; mem | grep airflow; procs
  echo "max_active_tasks limit lines: $(docker compose logs airflow | grep -c "max_active_tasks limit")"
}

try_limits 16 4; try_limits 4 8; try_limits 8 8


===== parallelism=16 max_active_tasks_per_dag=4
parallelism=16 max_active_tasks_per_dag=4
-- idle
afcon-airflow-1 1.074GiB / 2.344GiB 3.13%
workers = 16  sleeping tasks = 0
max running = 4
-- running
afcon-airflow-1 1.41GiB / 2.344GiB 10.73%
workers = 16  sleeping tasks = 4
max_active_tasks limit lines: 0

===== parallelism=4 max_active_tasks_per_dag=8
parallelism=4 max_active_tasks_per_dag=8
-- idle
afcon-airflow-1 1.022GiB / 2.344GiB 4.69%
workers = 4  sleeping tasks = 0
max running = 4
-- running
afcon-airflow-1 1.287GiB / 2.344GiB 6.20%
workers = 4  sleeping tasks = 4
max_active_tasks limit lines: 0

===== parallelism=8 max_active_tasks_per_dag=8
parallelism=8 max_active_tasks_per_dag=8
-- idle
afcon-airflow-1 1.039GiB / 2.344GiB 3.10%
workers = 8  sleeping tasks = 0
max running = 8
-- running
afcon-airflow-1 1.465GiB / 2.344GiB 5.79%
workers = 8  sleeping tasks = 8
max_active_tasks limit lines: 0

parallelism 16, max_active_tasks 4: parallelism을 8에서 16으로 두 배 올렸지만 running은 4 그대로예요. 워커만 16개로 늘었어요.
parallelism 4, max_active_tasks 8: 이번에는 parallelism 4에서 멈춰요.
둘 다 8: running이 8까지 올라가요.

running 최댓값은 매번 두 값 중 작은 쪽이었어요. 태스크가 돌 때 늘어난 메모리를 태스크 수로 나누면 16/4일 때 약 86MiB, 4/8은 약 68MiB, 8/8은 약 55MiB였어요.

정리하면 걸려 있는 한도를 올려야 태스크가 더 돌고, 다른 한도를 올리면 워커와 메모리만 늘어나요.

실습 3: pool과 slot으로 외부 자원을 지킵니다

예를 들어서 외부 API가 동시 요청 3개까지만 받는다고 해볼게요. 그러면 그 API를 부르는 태스크가 어느 DAG에 있든, 합쳐서 3개를 넘으면 안 되죠.

parallelism은 모든 태스크를 합친 수라 이 용도에 맞지 않고, max_active_tasks는 run 하나 안의 수예요. pool은 이름으로 묶기 때문에, DAG와 상관없이 같은 이름을 쓰는 태스크를 함께 셉니다.

dags/pools.py는 pool api를 함께 쓰는 DAG를 만들어요. pool_a와 pool_b는 slot 1칸짜리 태스크 5개씩이에요. 같은 파일의 pool_heavy(pool_slots=2인 태스크 3개)는 이 글에서는 쓰지 않아요.

실습 2에서 100초만 보고 둔 wide_100 run이 아직 돌고 있어요. 그래서 그 run이 워커를 차지하지 않게 DB를 새로 만들고, parallelism=8, max_active_tasks_per_dag=4로 돌아간 뒤 pool api를 slot 3개로 만들어 볼게요.

docker compose down -v 2> /dev/null
PAR=8 MAT=4 docker compose up -d 2> /dev/null
waitup
limits 8 4
af pools set api 3 "external api"
af pools list

parallelism=8 max_active_tasks_per_dag=4
Pool api created
pool         | slots | description  | include_deferred | team_name
=============+=======+==============+==================+==========
default_pool | 128   | Default pool | False            | None     
api          | 3     | external api | False            | None

pools list는 slot 수만 보여 주고, 지금 몇 개를 쓰는지는 보여 주지 않아요. 사용량은 pool_usage로 봅니다. 서로 다른 DAG인 pool_a와 pool_b를 함께 trigger해 볼게요. 태스크는 합쳐서 10개예요.

af dags trigger pool_a > /dev/null; af dags trigger pool_b > /dev/null
pool_usage api > out/5-pool-1.txt

time     dag_id     running queued scheduled slots_in_use
16:36:04 pool_a           3      0         2            3
16:36:04 pool_b           0      0         5            3

16:36:51 pool_a           0      0         2            0
16:36:51 pool_b           0      0         5            0
16:36:52 pool_a           2      0         0            3
16:36:52 pool_b           1      0         4            3

처음에는 pool_a가 slot 3개를 다 썼고, pool_b의 태스크 5개는 scheduled로 기다려요. 두 DAG 모두 run 하나의 max_active_tasks(4)보다 적게 돌고 있으니 pool에 걸린 거예요.
pool_a의 첫 3개가 끝나자 pool_a 2개와 pool_b 1개가 함께 들어갔어요. DAG가 달라도 slot 합은 3을 넘지 않습니다.

Pools 문서는 pool이 차면 "runnable tasks get queued and their state will show as such in the UI"라고 써요. 그런데 3.3.2에서 pool에 걸려 기다리는 태스크는 queued가 아니라 scheduled였어요. scheduler는 slot이 남은 pool의 태스크만 queued로 바꾸기 때문이에요.

정리

범위: Airflow 3.3.2(apache/airflow:3.3.2-python3.12), LocalExecutor, 메타데이터 DB PostgreSQL 17, Airflow 컨테이너 메모리 2.344GiB와 CPU 1.8개, 태스크는 모두 sleep 60.

parallelism 8, max_active_tasks 4에서 태스크 100개짜리 run의 running은 세 번 모두 4였어요. 동시 실행 수는 가장 작은 한도에서 멈추니, 걸린 한도부터 찾아보세요.
parallelism을 16으로 올려도 running은 4였고 워커만 16개로 늘었으며, 16/4, 4/8, 8/8에서 running 최댓값은 매번 두 값 중 작은 쪽이었어요. 도는 태스크는 하나에 55~86MiB였으니, 걸리지 않은 한도는 올리지 말고 한도를 올리기 전에 메모리를 계산해 보세요.
slot 3개짜리 pool api에서 서로 다른 DAG의 태스크는 합쳐 3개를 넘지 않았어요. 여러 DAG가 함께 쓰는 외부 자원은 pool로 묶어 주세요.
max_active_tasks와 pool에 걸린 태스크는 모두 scheduled로 남았어요(Pools 문서는 queued라고 써요). 이 글에 싣지 않은 추가 측정에서는 max_active_runs에 걸린 run과 LocalExecutor 워커가 찬 태스크가 queued로 남았으니, queued가 몇십 초씩 남으면 max_active_runs와 워커 수를 먼저 봐 주세요.
max_active_tasks에 걸렸다는 로그는 75번의 교대 동안 4줄이었어요. scheduled로 기다리면 run별 running과 queued 합을 max_active_tasks와, pool의 사용 slot을 slot 수와 나란히 비교해 보세요.