DataStream API와 상태(State) 심화

DataStream API는 무엇을 다루는가

Apache Flink의 DataStream API는 무한 스트림(unbounded stream)과 유한 배치(bounded stream)를 동일한 방식으로 처리하는 저수준 프로그래밍 모델이다. Table API / Flink SQL이 옵티마이저에 실행 계획을 맡긴다면, DataStream API는 개발자가 오퍼레이터 체인, 상태 구조, 타이머, 사이드 출력 을 직접 설계한다. 실시간 사기 탐지, 이벤트 패턴 감지, 커스텀 집계처럼 SQL 한 줄로 표현하기 어려운 복잡한 스트림 로직에 적합하다.

핵심 변환 연산자

DataStream API의 변환은 크게 세 범주로 나뉜다.

keyBy()는 DataStream을 KeyedStream 으로 바꾸는 분기점이다. 이 지점을 기준으로 Keyed State 를 사용할 수 있게 되며, 같은 키를 가진 레코드가 항상 동일한 서브태스크로 라우팅된다.

상태의 두 축: Keyed State vs Operator State

Keyed State 는 keyBy() 이후의 오퍼레이터에서만 사용할 수 있다. 키별로 완전히 격리된 상태 파티션을 갖기 때문에 동시성 문제 없이 단일 스레드에서 안전하게 읽고 쓸 수 있다.

Operator State 는 키와 무관하게 오퍼레이터 인스턴스 단위로 상태를 가진다. Kafka 소스가 자신이 담당하는 파티션의 오프셋을 기억하는 것이 대표적인 예다. 병렬도가 바뀌면 Flink는 인스턴스 간에 상태를 재분배한다.

Keyed State 다섯 가지 타입

Flink는 목적에 맞게 다섯 가지 Keyed State 타입을 제공한다.

RichFunction에서 상태 등록하기

상태를 직접 선언하는 가장 기본 방법은 RichFlatMapFunction이나 RichMapFunction에서 open() 메서드를 재정의하는 것이다. open()은 오퍼레이터 인스턴스가 초기화될 때 한 번만 호출되며, StateDescriptor를 통해 Flink 런타임에 상태를 등록한다.

public class FraudDetector extends RichFlatMapFunction<Transaction, Alert> {

    // 필드로 선언하되, open()에서 초기화
    private transient ValueState<Boolean> flagState;
    private transient ValueState<Long>    timerState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Boolean> flagDesc =
            new ValueStateDescriptor<>("flag", Types.BOOLEAN);
        flagState = getRuntimeContext().getState(flagDesc);

        ValueStateDescriptor<Long> timerDesc =
            new ValueStateDescriptor<>("timer", Types.LONG);
        timerState = getRuntimeContext().getState(timerDesc);
    }

    @Override
    public void flatMap(Transaction tx, Collector<Alert> out) throws Exception {
        Boolean lastFlagged = flagState.value();   // null이면 미설정
        if (lastFlagged != null && lastFlagged) {
            if (tx.getAmount() > 500.0) {
                out.collect(new Alert(tx.getAccountId()));
            }
            flagState.clear();
        }
        if (tx.getAmount() < 1.0) {
            flagState.update(true);    // 소액 거래 → 플래그 설정
        }
    }
}

transient 키워드: Java 직렬화에서 필드를 제외한다. Flink는 상태 자체를 직렬화하지 않고 오퍼레이터 인스턴스를 직렬화해 TaskManager에 전달하므로, 런타임 핸들인 ValueState를 transient로 선언해 오직 open()에서만 초기화한다.

KeyedProcessFunction과 타이머

RichFlatMapFunction이 상태만 다룬다면, KeyedProcessFunction은 타이머(Timer) 까지 제어할 수 있는 Flink DataStream API의 가장 강력한 저수준 도구다.

processElement()와 onTimer()는 같은 키의 상태를 공유한다. 타이머는 키와 타임스탬프 조합으로 중복 제거되며, 같은 타임스탬프에 여러 번 등록해도 onTimer()는 한 번만 실행된다.

public class FlagCleanupFunction
    extends KeyedProcessFunction<Long, Transaction, Alert> {

    private ValueState<Boolean> flagState;

    @Override
    public void open(Configuration conf) throws Exception {
        flagState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("flag", Boolean.class));
    }

    @Override
    public void processElement(Transaction tx,
                               Context ctx,
                               Collector<Alert> out) throws Exception {
        if (tx.getAmount() < 1.0) {
            flagState.update(true);
            // 1분 뒤 처리 시간 타이머 등록
            long timer = ctx.timerService().currentProcessingTime()
                         + Duration.ofMinutes(1).toMillis();
            ctx.timerService().registerProcessingTimeTimer(timer);
        } else if (Boolean.TRUE.equals(flagState.value())
                   && tx.getAmount() > 500.0) {
            out.collect(new Alert(tx.getCurrentKey()));
            flagState.clear();
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
                        Collector<Alert> out) throws Exception {
        // 1분이 지나도 대형 거래 없으면 플래그 해제
        flagState.clear();
    }
}

Broadcast State Pattern

스트림 처리에서 흔한 요구 중 하나는 설정·규칙 스트림을 모든 병렬 인스턴스에 공유 하는 것이다. Flink는 이를 위해 Broadcast State Pattern 을 제공한다.

예를 들어, 실시간 사기 탐지 규칙이 외부 시스템에서 주기적으로 갱신된다면 규칙 스트림을 브로드캐스트하고 주 데이터 스트림과 연결(connect)해 각 TaskManager 인스턴스에서 규칙을 공유할 수 있다.

// 규칙 스트림을 브로드캐스트 상태로 변환
BroadcastStream<Rule> ruleStream =
    rules.broadcast(ruleStateDescriptor);

// 주 데이터 스트림과 연결
DataStream<Alert> alerts = transactions
    .connect(ruleStream)
    .process(new RuleEvaluator());

RuleEvaluator는 BroadcastProcessFunction을 상속하며, 두 메서드를 구현한다. processElement()는 각 트랜잭션을 처리하면서 읽기 전용 브로드캐스트 상태에서 현재 규칙을 읽는다. processBroadcastElement()는 새 규칙이 도착할 때 브로드캐스트 상태를 갱신한다. 브로드캐스트 상태는 체크포인트에 포함되므로 재시작 후에도 규칙이 보존된다.

State TTL — 자동 만료와 정리

사용자 세션 추적이나 사기 감지 플래그처럼 일정 시간이 지나면 필요 없어지는 상태에는 TTL(Time-to-Live) 을 설정해야 한다. TTL 없이 상태를 방치하면 RocksDB 디스크나 힙 메모리가 무한히 늘어난다.

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Duration.ofHours(1))          // 만료 시간
    .setUpdateType(UpdateType.OnCreateAndWrite) // 쓸 때마다 TTL 갱신
    .setStateVisibility(StateVisibility.NeverReturnExpired) // 만료값 반환 금지
    .cleanupInRocksdbCompactFilter(1000, Duration.ofHours(1)) // RocksDB 컴팩션 청소
    .build();

ValueStateDescriptor<Boolean> flagDesc =
    new ValueStateDescriptor<>("flag", Boolean.class);
flagDesc.enableTimeToLive(ttlConfig);

TTL은 상태 저장 시 타임스탬프를 추가로 기록한다. RocksDB 백엔드에서는 값 하나당 8바이트 오버헤드가 생긴다. 만료된 상태는 즉시 삭제되는 것이 아니라 베스트에포트(best-effort) 로 정리된다. RocksDB 백엔드는 컴팩션 시 컴팩션 필터로 만료 항목을 제거하고, 힙 기반 백엔드는 증분 정리(incremental cleanup) 방식으로 읽기·쓰기 시에 순회하며 제거한다.

Flink 2.0의 State V2 API

Flink 2.0에서는 State V2 라는 새 비동기(async) 상태 API가 도입됐다. 기존 동기식 ValueState.value()는 상태 읽기가 완료될 때까지 스레드를 블로킹했지만, State V2는 asyncValue()처럼 CompletableFuture를 반환해 I/O 대기 중에도 다른 레코드를 처리할 수 있다. ForSt 백엔드(원격 스토리지를 1차 저장소로 쓰는 Flink 2.0 신규 백엔드)와 결합하면 네트워크 I/O 지연을 숨기면서 처리량을 높일 수 있다. 단, State V2 API는 아직 실험적(experimental) 단계다.

상태 크기 모니터링

상태가 예상 외로 커지면 TaskManager 메모리 부족이나 체크포인트 지연으로 이어진다. 실전에서는 다음 메트릭을 지속적으로 관찰한다.

• rocksdb.cur-size-all-mem-tables — RocksDB 메모리 테이블 크기
• flink_taskmanager_job_task_operator_numRecordsIn — 오퍼레이터별 처리 레코드 수
• 체크포인트 완료 시간 — 상태 크기 증가와 정비례

상태 폭증의 원인은 대부분 TTL 미설정, 무한 ListState 누적, Broadcast State에 오래된 규칙 방치 세 가지 중 하나다.

References

• https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream/fault-tolerance/state/
• https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/datastream-v2/context_and_state_processing/
• https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/state_v2/
• https://nightlies.apache.org/flink/flink-docs-stable/docs/learn-flink/event_driven/
• https://developer.confluent.io/courses/flink-java/working-with-keyed-state-in-flink/
• https://nightlies.apache.org/flink/flink-docs-master/docs/ops/state/state_backends/
• https://flink.apache.org/2019/05/17/state-ttl-in-flink-1.8.0-how-to-automatically-cleanup-application-state-in-apache-flink/
• https://streamkap.com/resources-and-guides/flink-state-management