Apache Airflow 3.3: 상태 저장 태스크(AIP-103)·다중 언어 SDK(AIP-108)·플러거블 재시도로 달라진 운영 기준

요약

2026년 7월 6일 출시된 Apache Airflow 3.3.0은 두 가지 구조적 변화를 담고 있다.

첫 번째는 태스크 상태 저장(AIP-103)이다. 이전까지 Airflow 태스크가 실행 간 데이터를 전달하려면 XCom에 전체 결과를 직렬화하거나 외부 저장소를 직접 관리해야 했다. 3.3.0은 task_state_store라는 공식 키-값 상태 접근자를 도입해 워터마크·진행 상태·고수위 마크 같은 운영 데이터를 Airflow 안에서 관리할 수 있게 했다.

두 번째는 다중 언어 태스크 SDK(AIP-108, 실험적)이다. DAG와 스케줄링은 여전히 파이썬이지만, 태스크 실행 로직은 이제 Java나 Go로 작성할 수 있다. Coordinator 레이어가 파이썬 DAG와 비파이썬 런타임을 중개한다.

핵심 요약:

• AIP-103 Stateful Tasks: task_state_store로 태스크 실행 간 상태 지속. SparkSubmitOperator는 ResumableJobMixin을 통해 이미 채택.
• AIP-108 Multi-Language SDK (실험적): @task.stub 선언 + Coordinator 라우팅으로 Java/Go 태스크 지원.
• AIP-105 Pluggable Retry Policies: 태스크별 커스텀 재시도 전략 연결.
• Asset 파티셔닝 확장: FanOutMapper, FixedKeyMapper, SegmentWindow 등 새 매퍼·윈도 추가.
• Breaking: OpenTelemetry 메트릭이 Gauge에서 Histogram으로 변경.

배경: Airflow 3.x의 진화 방향

Airflow 3.0(2026년 초)은 스케줄러·실행기 분리와 Asset 기반 데이터 인식 워크플로를 안정화했다. 3.1은 Human-in-the-loop 패턴(대기·승인 태스크)을, 3.2는 Asset 파티셔닝과 대규모 데이터 인식 워크플로를 강화했다.

3.3의 방향은 운영자가 현실에서 마주치는 두 가지 문제를 해결하는 것이다.

1. 상태 관리: Kafka 토픽이나 S3 버킷을 감시하는 태스크는 어디까지 처리했는지 기억해야 한다. XCom으로 워터마크를 넘기면 직렬화 오버헤드가 발생하고, 외부 DB에 저장하면 Airflow 외부 의존성이 늘어난다.
2. 언어 제약: 데이터 플랫폼 팀이 Java나 Go로 작성한 처리 로직을 Airflow에 통합하려면 파이썬 래퍼를 만들거나 BashOperator로 실행해야 했다. 타입 안전성, 변수·커넥션 주입, XCom 통신이 모두 수동이었다.

AIP-103: 상태 저장 태스크

설계 원칙

task_state_store는 태스크 컨텍스트에 바인딩된 키-값 저장소다. XCom과 다른 점은 세 가지다.

기본 사용 패턴

from airflow.decorators import task

@task
def process_kafka_topic(topic: str, **context):
    state = context["task_state_store"]

    # 이전 실행의 오프셋 읽기 (없으면 0으로 시작)
    last_offset = state.get("last_offset", default=0)

    # 처리 실행
    records = consume_from(topic, start_offset=last_offset)
    process(records)
    new_offset = last_offset + len(records)

    # 다음 실행을 위해 상태 저장
    state.set("last_offset", new_offset)
    state.set("last_processed_at", datetime.utcnow().isoformat())

    return new_offset

재시도가 발생해도 last_offset이 유지되므로 같은 레코드를 이중 처리하지 않는다. 성공 후에는 clear_on_success=True로 설정한 키만 삭제된다.

Asset 상태 저장

Asset(데이터 집합)도 고유한 상태 저장소를 갖는다.

from airflow.decorators import task
from airflow.sdk import Asset

raw_orders = Asset("s3://bucket/orders/raw/")

@task(outlets=[raw_orders])
def ingest_orders(**context):
    asset_state = context["asset_state_store"][raw_orders]

    # Asset 처리 진행 상태 저장
    last_file = asset_state.get("last_ingested_file")
    new_files = list_s3_files_after(last_file)

    for f in new_files:
        ingest(f)
        asset_state.set("last_ingested_file", f)

SparkSubmitOperator와 ResumableJobMixin

SparkSubmitOperator는 이미 ResumableJobMixin을 통해 AIP-103을 채택했다. 이전에는 Spark 잡이 시간 초과로 실패하면 처음부터 다시 제출해야 했다. 이제는 잡 ID를 상태에 저장하고 재시도 시 동일 잡에 재연결한다.

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

submit = SparkSubmitOperator(
    task_id="daily_etl",
    application="s3://jobs/daily_etl.py",
    # 재시도 시 같은 잡에 재연결 (ResumableJobMixin 동작)
    resumable=True,
)

AIP-108: 다중 언어 태스크 SDK (실험적)

Coordinator 아키텍처

AIP-108의 핵심은 Coordinator 레이어다. DAG는 파이썬으로 선언하되, 태스크 실행을 Java나 Go Coordinator에 위임한다.

from airflow.decorators import task

# DAG 안의 태스크 선언 — 실행 언어를 지정
@task.stub(queue="java-workers")
def transform_records(dataset_uri: str, schema_version: int) -> dict:
    """이 태스크의 실제 구현은 Java로 작성됨.
    파라미터 타입과 반환 타입은 파이썬 타입 힌트로 선언."""
    ...  # 실행 시 JavaCoordinator가 이 함수를 대신 실행

@task.stub(queue="go-workers")
def validate_output(result: dict) -> bool:
    """Go 바이너리로 구현된 검증 태스크."""
    ...

Airflow 워커는 queue 값을 보고 작업을 적절한 Coordinator로 라우팅한다.

JavaCoordinator 동작

JavaCoordinator는 JVM 위에서 실행되며 Airflow Execution API를 통해 Variables, Connections, XCom을 파이썬 측과 교환한다.

// Java 태스크 구현 (Airflow Java SDK)
@AirflowTask(taskId = "transform_records")
public class TransformRecordsTask implements AirflowTaskRunner {

    @Override
    public Map<String, Object> run(TaskContext context) {
        // Variables와 Connections를 Execution API로 자동 주입
        String jdbcUrl = context.getConnection("postgres_prod").getUri();
        String datasetUri = (String) context.getXcom("dataset_uri");

        // 실제 처리 로직
        DatasetTransformer transformer = new DatasetTransformer(jdbcUrl);
        Map<String, Object> result = transformer.transform(datasetUri);

        return result;  // 자동으로 XCom으로 반환
    }
}

ExecutableCoordinator — Go 바이너리

Go는 컴파일된 단일 바이너리로 배포된다. ExecutableCoordinator가 실행 중인 바이너리와 stdio 기반 프로토콜로 통신한다.

// Go 태스크 구현
package main

import (
    "encoding/json"
    airflow "github.com/apache/airflow-go-sdk"
)

func ValidateOutput(ctx *airflow.TaskContext) (bool, error) {
    var result map[string]interface{}
    ctx.XCom().Pull("transform_result", &result)

    // 검증 로직
    return isValid(result), nil
}

func main() {
    airflow.Run(ValidateOutput)
}

아키텍처 다이어그램: Airflow 3.3 다중 언어 태스크 흐름

AIP-105: 플러거블 재시도 정책

기존 Airflow는 재시도 횟수(retries)와 대기 시간(retry_delay)만 설정할 수 있었다. 특정 예외만 재시도하거나, 비즈니스 로직에 따라 재시도 여부를 동적으로 결정하는 것이 어려웠다.

AIP-105는 태스크별로 커스텀 재시도 정책 클래스를 연결할 수 있게 한다.

from airflow.models.retry_policy import BaseRetryPolicy
from airflow.decorators import task

class TransientErrorRetryPolicy(BaseRetryPolicy):
    """일시적 오류에만 재시도하는 정책."""

    RETRYABLE_EXCEPTIONS = (ConnectionError, TimeoutError, RateLimitError)
    MAX_RETRIES = 5

    def should_retry(self, exception: Exception, attempt: int) -> bool:
        if attempt >= self.MAX_RETRIES:
            return False
        return isinstance(exception, self.RETRYABLE_EXCEPTIONS)

    def retry_delay(self, attempt: int) -> int:
        # 지수 백오프: 1, 2, 4, 8, 16초
        return 2 ** attempt

@task(retry_policy=TransientErrorRetryPolicy())
def call_external_api(endpoint: str) -> dict:
    return requests.get(endpoint).json()

재시도 정책을 클래스로 분리하면 다수의 태스크에 동일 정책을 일관성 있게 적용하고 단독으로 테스트할 수 있다.

Asset 파티셔닝 확장

3.3은 Asset 파티셔닝 매퍼와 윈도 타입을 확장했다.

FanOutMapper를 활용하면 단일 수집 Asset이 업데이트될 때 지역별 처리 DAG를 동시에 트리거할 수 있다.

운영 고려사항

Breaking Change: OTel 메트릭 Gauge → Histogram

3.3.0은 OpenTelemetry 메트릭 타입을 일부 변경했다. Gauge로 수집하던 지연 시간 메트릭 일부가 Histogram으로 전환됐다.

영향 범위:

• Prometheus/Grafana 대시보드에서 해당 메트릭을 참조하는 패널 — PromQL 쿼리 갱신 필요.
• 알림 규칙에서 Gauge 기반 임계값 비교를 사용하는 경우 — Histogram 분위수(quantile) 기반으로 재작성.

# 변경 전 (Gauge)
- alert: AirflowTaskDurationHigh
  expr: airflow_task_duration > 300

# 변경 후 (Histogram 분위수)
- alert: AirflowTaskDurationHigh
  expr: histogram_quantile(0.95, airflow_task_duration_bucket) > 300

AIP-108 적용 전 체크리스트

• [ ] Java Coordinator: Airflow Java SDK 의존성 추가 (airflow-java-sdk >= 3.3.0)
• [ ] Go Coordinator: 바이너리를 워커 이미지에 포함 또는 사이드카로 마운트
• [ ] 큐 설정: java-workers, go-workers 큐를 별도 워커 풀로 분리
• [ ] Execution API 권한: Coordinator가 Airflow API에 접근할 수 있는 서비스 계정 설정
• [ ] 실험적 기능임을 팀 전체에 공유 — GA 이전에 API가 변경될 수 있음

task_state_store 운영 주의사항

• 상태 데이터는 Airflow Metadata DB에 저장된다. 워터마크처럼 크기가 작은 운영 메타데이터에 적합하다. 대용량 데이터를 직렬화하면 Metadata DB에 부하가 생긴다.
• 키별 보존 기간을 설정하지 않으면 영구 보존된다. 장기 운영 환경에서 주기적 정리 계획이 필요하다.
• clear_on_success=True를 설정하면 성공한 태스크의 해당 키가 삭제된다. 디버깅 목적으로 성공 후에도 상태를 보고 싶다면 이 옵션을 사용하지 말 것.

제한사항과 열린 질문

• AIP-108은 실험적: Java SDK와 ExecutableCoordinator 프로토콜은 GA 이전에 변경될 수 있다. 프로덕션 적용 시 버전 고정 권장.
• Metadata DB 부하: task_state_store의 저장 백엔드가 기본적으로 Metadata DB다. 고부하 환경에서는 별도 상태 백엔드 구성 필요 여부를 검토.
• 다중 언어 디버깅: Java/Go 태스크의 예외 스택이 Airflow UI에 어떻게 노출되는지 상세한 검증이 필요하다.

Open question: task_state_store의 커스텀 백엔드 인터페이스가 3.4에서 공개될지 로드맵에 명시되지 않았다. Kafka Streams 상태 저장소처럼 외부 KV 저장소(Redis, DynamoDB)를 백엔드로 사용하는 커뮤니티 요청이 있다.

References

• Apache Airflow 3.3.0 공식 블로그: https://airflow.apache.org/blog/airflow-3.3.0/
• GitHub 릴리스 노트: https://github.com/apache/airflow/releases/tag/3.3.0
• AIP-103 Task State Management: https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/406623137/AIP-103+Task+State+Management
• AIP-108 Multi-Language Task SDK (Confluence): https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-108
• AIP-105 Pluggable Retry Policies (Confluence): https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-105
• Apache Airflow 3.3.1 버그픽스 (2026-08-12): https://github.com/apache/airflow/releases/tag/3.3.1