LLM WikiAccess-protected knowledge portal

WIKI

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 라는 공식 키 값 상태 접근자를 도입해 워터마크·진행 상태·고수위 마크 같은 운영 데이터를 Ai

경로human/study/content/database-frontier/134-airflow-3-3-stateful-tasks-multilang-sdk.md
카테고리Study
태그#airflow #multilang #mysql #sdk #stateful #study #tasks

요약

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와 비파이썬 런타임을 중개한다.

핵심 요약:

배경: 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과 다른 점은 세 가지다.

특성XComtask_state_store
용도태스크 간 출력 전달태스크 자신의 운영 상태 저장
저장 시점태스크 완료 후태스크 실행 중 임의 시점
직렬화전체 값 직렬화키 단위 갱신
보존 정책DAG 실행 단위키별 보존 기간 설정 가능
성공 시 지우기지원 안 함clear_on_success=True 옵션

기본 사용 패턴

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 다중 언어 태스크 흐름

DAG (Python)
@task.stub 선언
스케줄·의존성 정의
↓
Scheduler
queue 기반
워커 라우팅
→
Python Worker
@task 함수
기존 파이썬 태스크
변경 없음
Java Worker
JavaCoordinator
JVM 런타임
Execution API 중개
Go Worker
ExecutableCoordinator
Go 바이너리
stdio 프로토콜
→
Execution API
Variables / Connections
XCom / task_state_store
↓
Metadata DB
상태·이력 저장
Airflow 3.3 — Coordinator 레이어 아키텍처

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 변경을 여러 파티션으로 팬아웃
FixedKeyMapper특정 키만 고정 파티션으로 라우팅
SegmentWindow레코드를 세그먼트 단위로 묶어 처리
시간 윈도 대기 정책파티션 완전 수신 전까지 다음 태스크 대기

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

운영 고려사항

Breaking Change: OTel 메트릭 Gauge → Histogram

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

영향 범위:

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

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

AIP-108 적용 전 체크리스트

task_state_store 운영 주의사항

제한사항과 열린 질문

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

References