Bong's Log

Apache Airflow를 단순히 DAG 작성 도구로만 접근하면 워크플로우 규모가 커질 때 스케줄러 지연(Scheduler lag), 메타스토어 병목, 좀비 태스크(Zombie task) 등 다양한 인프라 레벨의 장애에 직면하게 됩니다.

엔터프라이즈 환경에서 대용량 데이터 파이프라인과 RDBMS 마이그레이션 오케스트레이션을 안정적으로 운영하기 위해서는 Airflow를 구성하는 핵심 컴포넌트의 역할과 상호작용 매커니즘을 명확히 이해해야 합니다. Airflow의 5대 핵심 아키텍처와 내부 동작 흐름, 그리고 실무 운영 최적화 포인트를 정리합니다.

1. Airflow 5대 핵심 컴포넌트 구조

+-------------------------------------------------------------+
|                         Webserver                           |
|       (Flask UI, State Inspection, Trigger DAG/Task)        |
+-------------------------------------------------------------+
                               | Read / Write
+-------------------------------------------------------------+
|                    Metadata Database                        |
|       (MySQL / PostgreSQL - Single Source of Truth)         |
+-------------------------------------------------------------+
         ^                              ^
         | Read / Write                 | Read / Write
+-------------------+          +------------------------------+
|     Scheduler     |          |       Executor / Workers     |
| (DAG Parsing Loop)|--------->|  (Local / Celery / K8s)      |
| (Heartbeat Engine)| Queue/Msg|  (Task Instance Execution)   |
+-------------------+          +------------------------------+
         ^
         | (Deferred Task Triggering)
+-------------------+
|     Triggerer     |
|  (asyncio-based)  |
+-------------------+

2. 컴포넌트별 상세 동작 원리와 내부 매커니즘

(1) Scheduler: DAG Parsing Loop와 State Transition

Airflow Scheduler는 멀티프로세스로 구동되며 내부적으로 크게 두 가지 루프를 반복합니다.

• DAG Parsing Loop (DagFileProcessorProcess): dag_dir_list_interval 주기로 파이썬 파일들을 동적 파싱하여 메모리에 로드하고 Metadata DB의 dag, serialized_dag 테이블에 직렬화하여 저장합니다.
• Scheduling Loop: DB를 폴링하여 실행 가능한 상태(scheduled)의 TaskInstance를 식별하고, 의존성(Upstream, Trigger Rule)이 충족되면 queued 상태로 변경한 뒤 Executor로 전달합니다.

(2) Metadata Database: 트랜잭션과 격리 수준

Metadata DB는 시스템의 Single Source of Truth입니다.

• MySQL 구성 시 주의점: Airflow는 8.0 이상을 공식 지원하며, 기본 트랜잭션 격리 수준(REPEATABLE-READ)에서 데드락이 발생할 확률이 높으므로 READ COMMITTED 설정을 권장합니다.
• 주요 테이블: dag_run, task_instance, log, xcom, variable, serialized_dag.

(3) Executor: 분산 처리 엔진의 선택

• LocalExecutor: 단일 노드 멀티프로세스 환경에 적합. 소규모 배치 파이프라인이나 마이그레이션 전용 점프박스 인스턴스에 구축할 때 오버헤드가 적습니다.
• CeleryExecutor / KubernetesExecutor: 분산 워커 클러스터 환경. Celery는 Redis/RabbitMQ와 같은 브로커가 추가로 필요하며, K8sExecutor는 태스크마다 Pod를 생성하여 완전한 리소스 격리를 보장합니다.

(4) Triggerer: Deferrable Operator를 통한 리소스 효율화

전통적인 Sensor(예: SqlSensor, FileSensor)는 조건이 만족될 때까지 워커 슬롯 1개를 점유(worker slot holding)하여 리소스를 낭비합니다. Airflow 2.2+의 Triggerer는 Python asyncio 기반으로 태스크 상태를 DB에 deferred로 저장한 뒤 슬롯을 반환하고, 이벤트가 감지되었을 때만 다시 워커에 스케줄링합니다.

3. 태스크 라이프사이클과 상태 변화 흐름

[None] 
  │
  ▼ (Scheduler 식별)
[Scheduled] 
  │
  ▼ (Dependencies Checked & Queue Insert)
[Queued] 
  │
  ▼ (Worker Slot 할당 & 실행 시작)
[Running] ──(Deferrable Operator 트리거)──> [Deferred] ──(Event 완료)──┐
  │                                                                     │
  ├─────────────────────────────────────────────────────────────────────┘
  │
  ├───► [Success]
  │
  ├───► [Failed] ──► (Retries 남아있을 경우) ──► [Up_for_retry]
  │
  └───► [Upstream_failed]

태스크 실행 중 워커 프로세스가 갑자기 종료되거나 네트워크가 단절되면 메타스토어와 실제 프로세스 상태 간 불일치가 발생합니다. Scheduler는 주기적인 Heartbeat 점검을 통해 일정 시간(scheduler_zombie_task_threshold) 응답이 없는 태스크를 Zombie Task로 판정하고 failed 처리 후 재시도 루틴으로 넘깁니다.

4. 대규모 파이프라인을 위한 airflow.cfg 튜닝 포인트

대용량 ETL 및 DB 마이그레이션 작업 시 스케줄러 멈춤이나 DB Lock 병목을 방지하기 위한 핵심 파라미터 세팅입니다.

[core]
# 전체 Airflow 인스턴스에서 동시에 실행할 수 있는 최대 태스크 수
parallelism = 64

# DAG 파일 파싱 시 최상단 전역 코드가 무거울 경우 타임아웃 방지 (기본 30초 -> 60초)
dagbag_import_timeout = 60

# XCom 백엔드에 대용량 데이터를 올리지 않도록 주의 (필요시 커스텀 S3/GCS XCom 백엔드 구성)
max_active_tasks_per_dag = 16

[scheduler]
# DAG 폴더 파싱 주기(초). 파일이 많다면 30~60초로 늘려 Disk I/O 및 CPU 부하 경감
min_file_process_interval = 30
dag_dir_list_interval = 60

# 스케줄러 1회 루프당 메타 DB에서 큐로 넘길 TaskInstance 개수
max_tis_per_query = 32

# 워커 심장박동 점검 주기 및 좀비 감지 주기
scheduler_heartbeat_sec = 5
scheduler_zombie_task_threshold = 300

[database]
# 메타스토어 Connection Pool 관리 (SQLAlchemy 레벨)
sql_alchemy_pool_size = 20
sql_alchemy_max_overflow = 10
sql_alchemy_pool_recycle = 1800

5. DB 마이그레이션 DAG 모범 사례

스케줄러의 부하를 줄이기 위해서는 DAG 파일 내 Top-level(함수 밖)에서 DB 커넥션을 맺거나 무거운 계산을 수행하지 않아야 합니다.

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.mysql.hooks.mysql import MySqlHook

default_args = {
    'owner': 'data_platform',
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=3),
}

def extract_and_load_chunk(**context):
    """
    Top-level이 아닌 Task 실행 시점에 Hook을 호출하여
    Scheduler의 DAG 파싱 부하 및 DB Connection 누수를 차단합니다.
    """
    source_hook = MySqlHook(mysql_conn_id='source_legacy_db')
    target_hook = MySqlHook(mysql_conn_id='target_cloud_db')
    
    logical_date = context['logical_date']
    sql = """
        SELECT id, order_no, amount, created_at 
        FROM source_orders 
        WHERE created_at >= %s AND created_at < %s
    """
    start_dt = logical_date.strftime('%Y-%m-%d 00:00:00')
    end_dt = (logical_date + timedelta(days=1)).strftime('%Y-%m-%d 00:00:00')
    
    records = source_hook.get_records(sql=sql, parameters=(start_dt, end_dt))
    
    if records:
        target_hook.insert_rows(
            table='target_orders',
            rows=records,
            target_fields=['id', 'order_no', 'amount', 'created_at'],
            commit_every=5000
        )

with DAG(
    dag_id='migration_mysql_orders_chunk',
    default_args=default_args,
    description='Chunk-based daily partition migration pipeline',
    schedule_interval='@daily',
    start_date=datetime(2026, 1, 1),
    catchup=False,
    max_active_runs=3,  # 동시 실행 DAGRun 제한으로 Target DB I/O 보호
    tags=['migration', 'mysql', 'orders']
) as dag:

    migrate_task = PythonOperator(
        task_id='task_chunk_migration',
        python_callable=extract_and_load_chunk,
        pool='db_migration_pool',  # Concurrency Pool 제어로 Connection 고갈 방지
    )

아키텍처 관점의 핵심 체크리스트

• Top-level 코드 분리: DAG 파일의 루트 레벨에서 DB 조회나 API 호출을 실행하면, 스케줄러가 파싱할 때마다(수십 초 주기) 해당 코드가 실행되어 스케줄러 전체가 다운될 수 있습니다.
• Airflow Pool 활용: 타겟 DB의 max_connections 및 I/O 성능에 맞춰 Airflow UI에서 전용 Pool 슬롯 수를 지정하고 태스크에 할당하세요.
• XCom 남용 금지: 대용량 데이터를 태스크 간에 넘길 때 XCom을 사용하면 메타 DB의 용량이 폭증하고 락이 발생합니다. XCom은 실행 메타데이터(카운트, 파티션 키) 전달에만 국한해야 합니다.

오라클 데이터베이스 운영 환경에서 "Hot"과 "Cold"라는 키워드는 주로 두 가지 핵심 아키텍처 맥락에서 사용됩니다.

1. 백업 및 복구 아키텍처: 서비스 중단 유무에 따른 Hot Backup(온라인 백업) vs Cold Backup(오프라인 백업)
2. 메모리(SGA) 관리 아키텍처: Buffer Cache 내부의 Touch Count 기반 LRU 알고리즘(Hot End vs Cold End)

실무 현장에서 가장 빈번하게 다뤄지는 두 영역의 핵심 작동 원리와 내부 메커니즘,

그리고 반드시 알아야 할 운영 전략을 정리합니다.

1. 아키텍처 1: Hot Backup vs Cold Backup

백업 방식 비교 분석

Hot Backup의 내부 메커니즘: BEGIN BACKUP 시 발생하는 일

Hot Backup은 DB가 오픈되어 트랜잭션이 활발히 발생하는 도중에 OS 파일 복사(cp, dd 등)를 수행합니다. OS 블록 단위(보통 4KB~64KB)와 오라클 블록 단위(보통 8KB)의 불일치로 인해, OS가 파일을 읽어 복사하는 도중에 DBWR이 디스크 블록을 덮어쓰면 앞부분과 뒷부분의 SCN이 다른 Split Block(Fractured Block) 현상이 발생합니다.

오라클은 이를 방지하고 일관성을 보장하기 위해 특수한 모드를 지원합니다.

-- 테이블스페이스 단위 Hot Backup 시작
ALTER TABLESPACE USERS BEGIN BACKUP;

-- [OS 레벨 파일 복사 수행]
-- host cp /oradata/orcl/users01.dbf /backup/hotbackup/

-- 테이블스페이스 단위 Hot Backup 종료
ALTER TABLESPACE USERS END BACKUP;

ALTER TABLESPACE ... BEGIN BACKUP 실행 시 내부 동작

1. Checkpoint 강제 유발: 해당 테이블스페이스에 속한 데이터 파일들에 대해 체크포인트를 발생시켜, 메모리의 더티 블록(Dirty Block)을 디스크에 동기화합니다.
2. 데이터 파일 헤더 동결(Freeze): 데이터 파일 헤더의 Checkpoint SCN 갱신을 멈춥니다. 즉, DBWR이 데이터 파일 본문에는 계속 쓰기 작업을 수행하지만, 헤더의 체크포인트 카운터와 SCN은 백업 시작 시점으로 고정됩니다.
3. Full Image Block 로깅 (가장 중요한 오버헤드 요인):
   • 블록 분할(Fractured Block)로 백업본이 깨지는 문제를 해결하기 위해, 백업 모드 상태에서 최초로 변경되는 블록의 전체 이미지(Full Block Image)를 Redo Log에 기록합니다.
   • 평소에는 변경된 컬럼/로우 단위의 Change Vector만 기록하던 것과 달리, 블록 전체를 통째로 Redo Log에 쓰기 때문에 Redo Generation 량이 급증합니다.

💡실무 팁: Hot Backup의 Fracture Block 복구 원리

백업된 데이터 파일의 블록이 분할(Split)되어 깨져 있더라도, 복구(Recovery) 단계에서 아카이브 로그에 기록된 Full Image Block을 해당 블록 위에 덮어쓴 후, 이후 발생한 변경 사항(Change Vectors)을 롤포워드(Roll-Forward)하여 정합성을 완벽히 맞춥니다.

Hot Backup 실무 트러블슈팅: 백업 락(End Backup 누락)

가장 흔한 장애 케이스는 백업 스크립트 오류나 관리자 실수로 END BACKUP이 누락되는 경우입니다. 데이터 파일 헤더가 고정된 채 장시간 유지되면 대량의 Redo Log가 생성되어 디스크 풀(Archive Hang)을 유발하거나, 인스턴스 재기동 시 ORA-01113 에러가 발생합니다.

점검 쿼ㄹㅣ

SELECT f.file#, f.name, b.status, b.change#, b.time
FROM v$backup b
JOIN v$datafile f ON b.file# = f.file#
WHERE b.status = 'ACTIVE';

장애 조치

만약 DB가 다운된 상태에서 ORA-01113: file 1 needs media recovery now, ORA-01110: data file 1: '/oradata/users01.dbf' 에러와 함께 기동되지 않는다면 마운트 단계에서 백업 모드를 해제해야 합니다.

STARTUP MOUNT;

-- 활성화된 백업 상태 해제
ALTER DATABASE END BACKUP;

-- 데이터베이스 오픈
ALTER DATABASE OPEN;

2. 아키텍처 2: Buffer Cache 내부의 Hot / Cold LRU 알고리즘

오라클 SGA의 핵심 컴포넌트인 Database Buffer Cache에서도 메모리 효율을 극대화하기 위해 Hot/Cold 개념이 사용됩니다. 단순 LRU(Least Recently Used) 방식을 보완한 Touch Count 기반의 Mid-Point Insertion LRU 알고리즘이 바로 그것입니다.

       [ Cold End ] ---------------------- [ Hot End ]
           |                                   |
           v                                   v
    [LRU Replacement List]              [ 자주 사용되는 블록 유지 ]
           ^
           | (신규 로딩 블록 삽입 위치: Mid-point)

작동 원리

1. Mid-point Insertion: 디스크에서 새롭게 읽어온 버퍼 블록은 리스트의 최상위(Most Recently Used)가 아니라, Cold 영역의 헤드(Mid-point)에 위치합니다.
2. Touch Count 메커니즘:
   • 세션이 해당 블록을 엑세스(Pin)할 때마다 버퍼 헤더의 Touch Count(TCH)가 1씩 증가합니다.
   • 무분별한 래치 경합과 카운트 급증을 막기 위해 최소 3초(_db_aging_touch_time) 동안은 추가 터치를 무시합니다.
3. Hot End 승격: Touch Count가 임계치(기본값 2) 이상이 되면 해당 블록은 Hot End 영역으로 이동하여 캐시 방출 대상에서 제외됩니다.
4. 대량 스캔(Full Table Scan)의 캐시 오염 방지:
   • 대용량 테이블 스캔 시 메모리가 순식간에 밀려나는 문제를 막기 위해, FTS로 읽어들인 블록은 즉시 Cold 영역의 끝(LRU Tail)에 배치되어 가장 먼저 Aging-Out 되도록 유도합니다 (CACHE 힌트나 속성이 없는 기본 동작).

Buffer Cache 상태 모니터링 쿼리

SELECT 
    o.object_name,
    o.object_type,
    bh.status,
    bh.tch AS touch_count,
    COUNT(*) AS buffer_count
FROM v$bh bh
JOIN dba_objects o ON bh.objd = o.data_object_id
WHERE o.owner = 'SCOTT'
GROUP BY o.object_name, o.object_type, bh.status, bh.tch
ORDER BY touch_count DESC;

요약 및 운영 권고사항

• 백업 측면:
  • 사용자 스크립트 기반의 ALTER TABLESPACE BEGIN/END BACKUP은 Redo 로그 급증 및 휴먼 에러 위험이 큽니다.
  • 블록 변경 추적(BCT) 및 Split Block 문제를 내부 엔진 레벨에서 자동 처리하는 RMAN(Recovery Manager) 사용이 표준 권장사항입니다.
• 메모리(LRU) 측면:
  • 자주 조회되는 마스터 테이블은 KEEP Buffer Pool(ALTER TABLE ... STORAGE (BUFFER_POOL KEEP))을 할당하여 Cold 영역으로의 밀려남을 원천 차단하십시오.
  • 주기적으로 대용량 FTS를 유발하는 배치 쿼리는 버퍼 캐시 오염을 막기 위해 실행 계획 튜닝 또는 DIRECT PATH READ 유도를 검토해야 합니다.

Apache Airflow는 데이터 엔지니어링 생태계에서 사실상 표준 워크플로 오케스트레이션 도구로 자리 잡았습니다. 단순한 파이프라인 관리 도구로 접근했다가, DAG 수가 수백 개를 넘어서는 순간 스케줄러 랙(Scheduler Lag), Executor 아키텍처 한계로 인한 리소스 낭비, 메타데이터 DB 커넥션 고갈, 좀비 태스크(Zombie Tasks) 등 운영 장애를 겪는 조직이 많습니다.

Airflow의 핵심 내부 아키텍처를 해부하고, 프로덕션 환경에서 가장 고민이 깊은 Celery vs Kubernetes Executor 아키텍처 분석, 실무 모범 DAG 패턴, 그리고 메타스토어(MySQL) 최적화 전략을 정리합니다.

1. Airflow 아키텍처 심층 분석: 스케줄러 루프(Scheduler Loop)의 내부

Airflow의 핵심은 스케줄러(Scheduler)입니다. 스케줄러가 DAG 파일을 파싱하고 태스크 인스턴스를 큐(Queue)로 밀어 넣는 라이프사이클을 이해하지 못하면 성능 튜닝은 불가능합니다.

[DAG Directory] 
       │ (1) DagFileProcessorManager (멀티프로세스 파싱)
       ▼
 [Metastore (MySQL/PostgreSQL)] ── (2) SchedulerJobLoop (상태 확인: queued/running)
       ▲                                 │
       │ (상태 업데이트)                     ▼ (3) Executor (Celery / Kubernetes)
       └────────────────────────── [Worker Task Execution]

핵심 프로세스 라이프사이클

1. DAG Parsing Loop (DagFileProcessorProcess):
2. 백그라운드에서 주기적으로 .py 파일을 파이썬 인터프리터로 읽어 들여 직렬화(Serialization)한 뒤 메타데이터 DB의 dag_code, serialized_dag 테이블에 동기화합니다.
3. Scheduler Loop (_do_scheduling):
   • dag_run 테이블에서 실행 대상인 DagRun을 탐색합니다.
   • 선행 태스크의 완료 여부 및 트리거 규칙(trigger_rule)을 평가합니다.
   • 실행 가능한 태스크 인스턴스를 SCHEDULED → QUEUED 상태로 전환하고 Executor 버퍼로 전달합니다.
4. Task Execution Loop (Executor):
   • 워커가 작업을 수신해 실행 상태(RUNNING)로 바꾼 뒤, 완료 시 최종 상태(SUCCESS/FAILED)를 메타스토어에 반영합니다.

실무 주의점: DAG 파일의 최상위 스코프(Top-level scope)에 외부 API 호출, 무거운 DB 쿼리, 대용량 연산 로직을 작성하면 DAG 파싱 루프가 지연되어 전체 스케줄링 랙이 기하급수적으로 증가합니다.

2. Executor 아키텍처 비교: CeleryExecutor vs KubernetesExecutor

대규모 분산 환경을 설계할 때 가장 먼저 맞닥뜨리는 의사결정은 Executor 선정입니다. 각 아키텍처의 내부 구동 방식과 리소스 격리 메커니즘을 비교해 봅니다.

[ CeleryExecutor Architecture ]
Airflow Scheduler ──▶ Redis / RabbitMQ (Broker) ──▶ Static Celery Workers (Pre-warmed Pods)
                          ▲
Airflow Webserver ────────┴──▶ Metastore (Result Backend / MySQL)

[ KubernetesExecutor Architecture ]
Airflow Scheduler ──▶ K8s API Server ──▶ Dynamic Pod Spawn (Task 1 Pod)
                                      ──▶ Dynamic Pod Spawn (Task 2 Pod)
                                          (Pod Terminated after execution)

1) 아키텍처 및 내부 구동 원리 비교

• CeleryExecutor:
  • 동작 원리: 별도의 메시지 브로커(Redis 또는 RabbitMQ)를 두며, 상시 기동 중인 워커 노드(Static Workers)의 프로세스 풀(Worker Concurrency)에 Celery 큐를 통해 작업을 분배합니다.
  • 장점: 워커 프로세스가 이미 예열(Pre-warmed)되어 있으므로 Task 시작 오버헤드(Latency)가 수백 밀리초(ms) 수준으로 극히 낮습니다. 초 단위 또는 분 단위로 대량 실행되는 경량 태스크 파이프라인에 최적입니다.
  • 단점: 워커 환경이 정적이므로 서로 다른 DAG 간 Python 라이브러리 충돌(Dependency Hell)이 발생하기 쉽고, 야간 배치 몰림 현상 시 유휴 자원 낭비 또는 워커 부족(Queue Stalling)이 일어납니다.
• KubernetesExecutor:
  • 동작 원리: 메시지 브로커가 필요 없습니다. Airflow 스케줄러가 K8s API Server와 직접 통신(KubeClient)하여 각 태스크 인스턴스마다 독립된 Kubernetes Pod을 동적으로 생성하고, 작업 완료 시 Pod을 삭제합니다.
  • 장점: 완벽한 리소스 및 환경 격리를 제공합니다. 태스크마다 서로 다른 Docker 이미지, CPU/Memory 리소스 Request/Limit을 개별 할당할 수 있어 의존성 충돌과 OOM 전파 위험이 원천 차단됩니다.
  • 단점: 매 태스크마다 Pod 스케줄링, 이미지 풀(Pull), 컨테이너 초기화 단계를 거치므로 수 초에서 수십 초의 기동 오버헤드(Cold Start Latency)가 발생합니다.

2) Celery vs Kubernetes 상세 비교

3) 실무 선택 기준 및 하이브리드 대안 (CeleryKubernetesExecutor)

• KubernetesExecutor를 선택해야 하는 경우:
  1. 머신러닝, Spark 제출 등 태스크마다 요구하는 라이브러리 버전이나 OS 패키지가 판이할 때
  2. 특정 태스크가 일시적으로 수십 GB의 메모리를 사용하여 인접 태스크의 동반 장애(OOM-Killer)를 방지해야 할 때
  3. 클러스터 오토스케일러(Karpenter, Cluster Autoscaler)와 연계해 클라우드 비용을 유연하게 제어하고자 할 때
• CeleryExecutor를 선택해야 하는 경우:
  1. 수천 개의 태스크가 1~2분 간격으로 빠르게 트리거되어 Pod 기동 지연(5~10초)을 감당하기 어려울 때
  2. K8s API 서버의 초당 요청 한도(Throttling) 및 Watcher 연결 부하를 스케줄러에서 덜어내야 할 때

아키텍트 추천 (CeleryKubernetesExecutor):

Airflow는 두 환경의 장점을 합친 **CeleryKubernetesExecutor**를 지원합니다. 빈번하고 가벼운 정형 SQL/데이터 이동 배치는 Celery 큐로 밀어 즉각 실행하고, 고용량 메모리/특수 라이브러리가 필요한 작업은 태스크 레벨에서 queue="kubernetes"로 지정하여 동적 Pod으로 띄우는 것이 엔터프라이즈 환경의 정석 설계입니다.

3. 실무 Best Practice: TaskFlow API와 동적 태스크 매핑

Airflow 2.x 이상에서는 Python Decorator 기반의 TaskFlow API와 Airflow 2.3+의 Dynamic Task Mapping을 사용하는 것이 표준입니다.

from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.providers.mysql.hooks.mysql import MySqlHook

default_args = {
    "owner": "data_architect",
    "depends_on_past": False,
    "retries": 2,
    "retry_delay": timedelta(minutes=3),
    "execution_timeout": timedelta(minutes=30),
}

@dag(
    dag_id="mysql_partition_batch_pipeline",
    start_date=datetime(2026, 1, 1),
    schedule="0 2 * * *",  # 매일 새벽 2시 UTC
    catchup=False,
    max_active_runs=1,
    default_args=default_args,
    tags=["etl", "mysql", "production"],
)
def mysql_partition_pipeline():

    @task
    def extract_target_partitions() -> list[str]:
        """추출 대상 파티션 키 목록 동적 조회"""
        hook = MySqlHook(mysql_conn_id="source_mysql_prod")
        sql = """
            SELECT PARTITION_NAME 
            FROM information_schema.PARTITIONS 
            WHERE TABLE_SCHEMA = 'production' 
              AND TABLE_NAME = 'user_activity_logs'
              AND PARTITION_NAME IS NOT NULL
            ORDER BY PARTITION_ORDINAL_POSITION DESC 
            LIMIT 7;
        """
        records = hook.get_records(sql)
        return [record[0] for record in records]

    @task
    def process_partition(partition_name: str) -> dict:
        """각 파티션별 독립 처리 (동적 매핑 적용)"""
        hook = MySqlHook(mysql_conn_id="source_mysql_prod")
        query = f"""
            SELECT COUNT(*) AS row_cnt, MAX(updated_at) AS latest_sync
            FROM production.user_activity_logs PARTITION ({partition_name})
        """
        result = hook.get_first(query)
        return {"partition": partition_name, "count": result[0], "latest_sync": str(result[1])}

    @task
    def aggregate_summary(summaries: list[dict]):
        """배치 집계 결과 검증 및 적재"""
        total_rows = sum(item["count"] for item in summaries)
        print(f"[METRIC] Total Processed Rows Across Partitions: {total_rows}")

    # DAG 파이프라인 의존성 정의
    partitions = extract_target_partitions()
    mapped_results = process_partition.expand(partition_name=partitions)
    aggregate_summary(mapped_results)

mysql_partition_pipeline()

4. 메타스토어 병목 진단: MySQL 환경에서의 인덱스 및 커넥션 최적화

Airflow 메타스토어로 MySQL(또는 InnoDB 엔진)을 운영할 때 태스크 동시 실행 수가 늘어나면 메타스토어 락(Lock) 경합과 커넥션 고갈이 발생합니다. 특히 task_instance, dag_run, job 테이블이 주요 병목 대상입니다.

1) 주요 경합 쿼리와 커버링 인덱스(Covering Index)

스케줄러는 실행 대상을 찾기 위해 task_instance 테이블을 매 초 수십 회 폴링합니다.

-- 스케줄러가 실행 대상을 찾기 위해 빈번하게 날리는 쿼리
EXPLAIN SELECT task_id, dag_id, execution_date, state, try_number
FROM task_instance
WHERE state IN ('queued', 'scheduled')
  AND dag_id = 'mysql_partition_batch_pipeline';

• 진단: 실행 계획(EXPLAIN) 확인 시 type: ALL 또는 Using filesort가 관측된다면 인덱스 스캔을 타지 못하는 상태입니다.
• 튜닝 방안: (state, dag_id, execution_date) 복합 인덱스를 점검하고, 인덱스 컨디션 푸시다운(ICP) 및 커버링 인덱스가 동작하도록 유도해야 합니다.

2) InnoDB 파라미터 및 연결 풀 설정

Airflow 전용 MySQL 인스턴스는 쓰기/상태 갱신이 빈번한 고전적인 OLTP 특성을 띱니다.

5. 실무 핵심: airflow.cfg 성능 최적화 가이드

대규모 환경에서 스케줄러와 파서의 리소스 낭비를 줄이는 핵심 파라미터 튜닝입니다.

[core]
# 메타스토어 커넥션 풀 제어 (SQLAlchemy)
sql_alchemy_pool_size = 20
sql_alchemy_max_overflow = 10
sql_alchemy_pool_recycle = 1800

# 병렬 실행 제한 (전체 인스턴스 기준)
parallelism = 64
max_active_tasks_per_dag = 16
max_active_runs_per_dag = 3

[scheduler]
# DAG 파싱 빈도 조절 (초 단위) - 파일 변경이 잦지 않은 프로덕션은 수치를 늘릴 것
min_file_process_interval = 60
dag_dir_list_interval = 300

# 스케줄러 멀티프로세싱 스레드 수 (vCPU 코어 수에 맞춰 조절)
parsing_processes = 4

# 한 번의 쿼리 루프에서 스케줄러가 가져올 최대 태스크 인스턴스 수
max_tis_per_query = 512

6. 프로덕션 트러블슈팅: 좀비 태스크(Zombie Task)와 DB 클린업

1) Zombie Task 발생 원인과 방지책

• 증상: Airflow 웹 UI에는 상태가 계속 running으로 머물러 있으나, 실제 K8s Pod 또는 Celery Worker 프로세스는 OOM이나 네트워크 단절로 비정상 종료된 상태입니다.
• 메커니즘: 스케줄러가 메타데이터의 job 테이블에 주기적으로 하트비트(latest_heartbeat)를 갱신하지 못하는 태스크 인스턴스를 탐색하여 강제로 실패 처리합니다.
• 해결책:
  • scheduler_zombie_task_threshold: 기본 300초에서 워커 환경 특성에 맞춰 조정합니다.
  • 메모리 사용량이 예측 불가능한 대용량 데이터 전처리는 Airflow 워커 프로세스 내부에서 직접 돌리지 말고, KubernetesPodOperator를 통해 독립 컨테이너로 격리 위임해야 합니다.

2) 메타스토어 하우스키핑 (DB Archiving)

task_instance, log, xcom 테이블 데이터가 수백만 건 이상 누적되면 스케줄러의 SELECT 쿼리 속도가 급격히 저하됩니다. 정기적인 아카이빙 배치(airflow db clean CLI) 구축이 필수적입니다.

# 60일 이전의 메타데이터 보관 주기 정리 (Dry-run 검증 후 실행)
airflow db clean --clean-before-timestamp "$(date -d '60 days ago' +%Y-%m-%d)" --skip-archive -y