컨텍스트와 상태 처리

컨텍스트와 상태 처리 (Context and State Processing)

DataStream API V2에서 컨텍스트(context)와 상태(state)를 처리하는 방법을 설명하는 문서예요.

출처: 문서

본문

참고: DataStream API V2는 기존 DataStream API를 점진적으로 대체하기 위한 새로운 API 집합이에요. 현재 실험 단계(experimental stage)에 있으며 생산(production) 환경에서 완전히 사용 가능한 상태는 아니에요.

컨텍스트 (Context)

프로세스 연산의 이름 같은 속성과 달리, 어떤 정보(예: 현재 키)는 프로세스 함수가 실행될 때만 얻을 수 있어요. 프로세스 함수와 실행 엔진 사이의 다리를 구축하기 위해, DataStream API는 Runtime Context라고 불리는 통합 진입점(entrypoint)을 제공해요.

모든 컨텍스트 정보를 기능에 따라 여러 부분으로 나눌 수 있어요:

• JobInfo: 작업 이름, 실행 모드 등 작업과 관련된 모든 정보를 담아요.
• TaskInfo: 병렬도 등 태스크와 관련된 모든 정보를 담아요.
• MetricGroup: 메트릭 등록과 같은 메트릭 관련 컨텍스트를 관리해요.
• State Manager: 특정 상태에 접근하는 등 상태 관련 컨텍스트를 관리해요.
• Watermark Manager: 워터마크 트리거 등 워터마크 관련 컨텍스트를 관리해요.
• ProcessingTime Manager: 현재 프로세싱 타임을 얻는 것 등 프로세싱 타이머 관련 컨텍스트를 관리해요.

이런 컨텍스트 정보는 두 범주로 분류할 수 있어요:

• NonPartitionedContext: JobInfo, TaskInfo, MetricGroup, WatermarkManager를 포함해요.
• PartitionedContext: StateManager, ProcessingTimeManager를 포함해요.

일반적으로 PartitionedContext는 특정 파티션 내에서 데이터 처리가 기대될 때 ProcessFunction에 제공돼요. 예를 들어, 새 레코드 수신이나 타이머 트리거 시에요. 반면에 NonPartitionedContext는 특정 파티션이 관련 없을 때 제공돼요. 예를 들어, ProcessFunction의 초기화 또는 정리(clean-up) 시에요.

다음 코드 스니펫은 프로세스 함수의 병렬도와 실행 모드를 얻는 방법을 보여줘요:

new OneInputStreamProcessFunction<String, String>(){
    private transient int parallelism;
    
    private transient ExecutionMode executionMode;
    @Override
    public void open(NonPartitionedContext<String> ctx) throws Exception {
        parallelism = ctx.getTaskInfo().getParallelism();
        executionMode = ctx.getJobInfo().getExecutionMode();
    }
}

상태 처리 (State Processing)

상태는 상태를 가진 계산(stateful computation)의 기반이에요. DataStream API는 프로세스 함수에서 상태를 선언하고, 데이터 처리 중 상태에 접근하고 갱신하는 것을 지원해요.

일반적으로 "먼저 선언하고, 나중에 사용한다(declare first, use later)"는 원칙을 따라야 해요. 일반적으로 상태를 가진 프로세스 함수를 작성하는 것은 세 단계로 나뉘어요:

1. StateDeclaration의 형태로 상태를 정의한다.
2. ProcessFunction#usesStates에서 상태를 선언한다.
3. StateManager를 통해 상태를 얻고 갱신한다.

더 나아가기 전에, 상태를 가진 프로세스 함수가 어떻게 생겼는지 살펴볼게요:

private static class StatefulFunction implements OneInputStreamProcessFunction<Long, Long> {
    // Step1: Defining the state in the form of `StateDeclaration`.
    static final StateDeclaration.ListStateDeclaration<Long> LIST_STATE_DECLARATION =
            StateDeclarations.listStateBuilder("example-list-state", TypeDescriptors.LONG).build();
     
    // Step2: Declaring the state in `ProcessFunction#usesStates`
    @Override
    public Set<StateDeclaration> usesStates() {
        return Collections.singleton(LIST_STATE_DECLARATION);
    }
    
    @Override
    public void processRecord(Long record, Collector<Long> output, RuntimeContext ctx)
            throws Exception {
        // Step3: Getting and updating state via `StateManager`.
        ListState<Long> state =
                ctx.getStateManager().getState(LIST_STATE_DECLARATION);
        // do something with this state. For example, update the state by this record.
        state.update(Collections.singletonList(record));
    }
}

상태 정의 (Define State)

StateDeclaration은 특정 상태를 정의하는 데 사용되며, 상태 선언에는 두 가지 유형의 정보가 필요해요:

• Name: 상태의 고유 식별자로 사용돼요.
• RedistributionMode: 상태가 서로 다른 파티션들 사이에서 어떻게 재분배(redistribute)되는지 정의해요. 키처리 파티션 스트림(keyed partition stream)의 경우, 상태가 파티션 내에 경계지어져 있으므로 재분배가 필요 없어요. 하지만 논키 파티션 스트림(non-keyed partition stream)의 경우, 파티션이 병렬도에 따라 변하므로 상태가 어떻게 재분배할지 정의해야 해요.

선택할 수 있는 세 가지 RedistributionMode가 있어요:

• NONE: 재분배를 지원하지 않음.
• REDISTRIBUTABLE: 이 상태는 서로 다른 파티션 사이에서 안전하게 재분배될 수 있고, 구체적인 재분배 전략은 상태 자체가 결정해요.
• IDENTICAL: 상태는 서로 다른 파티션에서 동일함이 보장되므로, 재분배는 문제가 되지 않아요.

현재 다섯 가지 유형의 StateDeclaration가 있어요: ValueStateDeclaration, ListStateDeclaration, MapStateDeclaration, ReducingStateDeclaration, AggregatingStateDeclaration, BroadcastStateDeclaration. 이들은 각각 ValueState, ListState, MapState, ReducingState, AggregatingState, BroadcastState를 설명해요.

• ValueState: 갱신하고 검색할 수 있는 하나의 값을 유지해요. 값은 update(T)로 설정하고 T value()로 검색할 수 있어요.
• ListState: 요소 목록을 유지해요. 요소를 추가하고 현재 저장된 모든 요소에 대한 Iterable을 검색할 수 있어요. 요소는 add(T) 또는 addAll(List)로 추가되고, Iterable은 Iterable get()으로 검색할 수 있어요. 기존 목록을 update(List)로 덮어쓸 수도 있어요.
• ReducingState: 상태에 추가된 모든 값의 집계(aggregation)를 나타내는 단일 값을 유지해요. 인터페이스는 ListState와 비슷하지만 add(T)로 추가된 요소는 지정된 ReduceFunction으로 집계(aggregate)로 줄여져요.
• AggregatingState: 상태에 추가된 모든 값의 집계를 나타내는 단일 값을 유지해요. ReducingState와 달리 집계 타입은 상태에 추가되는 요소의 타입과 다를 수 있어요. 인터페이스는 ListState와 동일하지만 add(IN)로 추가된 요소는 지정된 AggregateFunction으로 집계돼요.
• MapState: 매핑 목록을 유지해요. 키-값 쌍을 상태에 넣고 현재 저장된 모든 매핑에 대한 Iterable을 검색할 수 있어요. 매핑은 put(UK, UV) 또는 putAll(Map)으로 추가돼요. 사용자 키와 연관된 값은 get(UK)로 검색할 수 있어요. 매핑, 키, 값에 대한 iterable 뷰는 각각 entries(), keys(), values()로 검색할 수 있어요. isEmpty()로 이 맵에 키-값 매핑이 있는지 확인할 수도 있어요.
• BroadcastState: BroadcastStream의 상태를 저장하기 위해 생성될 수 있어요. 이 상태는 동일한 요소가 프로세스 함수의 모든 인스턴스로 전송된다고 가정해요. 키-값 쌍을 상태에 넣고 현재 저장된 모든 매핑에 대한 Iterable을 검색할 수 있어요.

사용 편의를 위해 DataStream API는 StateDeclarations라는 보조 클래스를 제공하는데, 다양한 StateDeclaration 인스턴스를 생성하는 일련의 메서드를 캡슐화해요.

예를 들어:

// create a value state declaration
ValueStateDeclaration<Integer> valueStateDeclaration = StateDeclarations.valueState("example-value-state", TypeDescriptors.LONG);

// create a map state declaration
MapStateDeclaration<Long, String> mapStateDeclaration = StateDeclarations.mapState(
        "example-map-state", TypeDescriptors.LONG, TypeDescriptors.STRING);

// create a reducing state declaration with a `sum` reduce function
ReducingStateDeclaration<Long> reducingStateDeclaration =
        StateDeclarations.reducingState("example-reducing-state", TypeDescriptors.LONG, Long::sum);

상태의 타입 정보를 제공하기 위해, 위 예시에서 TypeDescriptors 클래스는 INT, LONG, BOOLEAN, STRING, LIST, MAP 등 일반적인 타입에 대한 사전 정의된 타입 디스크립터 집합을 제공해요. 타입이 사전 정의된 목록에 없으면, TypeDescriptor 인터페이스를 구현하는 클래스를 제공해 자신만의 타입 디스크립터를 정의할 수 있어요.

상태 선언 (Declare State)

앞서 말했듯이, DataStream API에서 상태를 사용할 때는 "먼저 선언하고, 나중에 사용한다"는 원칙을 따라야 해요. ProcessFunction에는 상태를 사전에 명시적으로 선언하는 useStates라는 메서드가 있어요.

default Set<StateDeclaration> usesStates() {
    return Collections.emptySet();
}

상태를 가진 함수의 경우 이 메서드를 오버라이드해야 하며, 각 특정 상태를 사용하기 전에 이 메서드에서 선언해야 해요. 이것은 Flink에게 함수 실행을 최적화할 더 많은 기회를 제공해요.

상태 얻기와 갱신 (Get and Update State)

StateManager는 프로세스 함수의 상태를 관리하는 데 사용돼요. 상태를 얻는 메서드와 현재 키를 검색하는 메서드를 제공해요.

public interface StateManager {
    /**
     * Get the key of current record.
     *
     * @return The key of current processed record.
     * @throws UnsupportedOperationException if the key can not be extracted for this function, for
     *     instance, get the key from a non-keyed partition stream.
     */
    <K> K getCurrentKey() throws UnsupportedOperationException;

    /**
     * Get the optional of the specific list state.
     *
     * @param stateDeclaration of this state.
     * @return the list state corresponds to the state declaration, this may be empty.
     */
    <T> Optional<ListState<T>> getStateOptional(ListStateDeclaration<T> stateDeclaration)
            throws Exception;

    /**
     * Get the specific list state.
     *
     * @param stateDeclaration of this state.
     * @return the list state corresponds to the state declaration
     * @throws RuntimeException if the state is not available.
     */
    <T> ListState<T> getState(ListStateDeclaration<T> stateDeclaration) throws Exception;

    /**
     * Get the optional of the specific value state.
     *
     * @param stateDeclaration of this state.
     * @return the value state corresponds to the state declaration, this may be empty.
     */
    <T> Optional<ValueState<T>> getStateOptional(ValueStateDeclaration<T> stateDeclaration)
            throws Exception;

    /**
     * Get the specific value state.
     *
     * @param stateDeclaration of this state.
     * @return the value state corresponds to the state declaration.
     * @throws RuntimeException if the state is not available.
     */
    <T> ValueState<T> getState(ValueStateDeclaration<T> stateDeclaration) throws Exception;

    /**
     * Get the optional of the specific map state.
     *
     * @param stateDeclaration of this state.
     * @return the map state corresponds to the state declaration, this may be empty.
     */
    <K, V> Optional<MapState<K, V>> getStateOptional(MapStateDeclaration<K, V> stateDeclaration)
            throws Exception;

    /**
     * Get the specific map state.
     *
     * @param stateDeclaration of this state.
     * @return the map state corresponds to the state declaration.
     * @throws RuntimeException if the state is not available.
     */
    <K, V> MapState<K, V> getState(MapStateDeclaration<K, V> stateDeclaration) throws Exception;

    /**
     * Get the optional of the specific reducing state.
     *
     * @param stateDeclaration of this state.
     * @return the reducing state corresponds to the state declaration, this may be empty.
     */
    <T> Optional<ReducingState<T>> getStateOptional(ReducingStateDeclaration<T> stateDeclaration)
            throws Exception;

    /**
     * Get the specific reducing state.
     *
     * @param stateDeclaration of this state.
     * @return the reducing state corresponds to the state declaration.
     * @throws RuntimeException if the state is not available.
     */
    <T> ReducingState<T> getState(ReducingStateDeclaration<T> stateDeclaration) throws Exception;

    /**
     * Get the optional of the specific aggregating state.
     *
     * @param stateDeclaration of this state.
     * @return the aggregating state corresponds to the state declaration, this may be empty.
     */
    <IN, ACC, OUT> Optional<AggregatingState<IN, OUT>> getStateOptional(
            AggregatingStateDeclaration<IN, ACC, OUT> stateDeclaration) throws Exception;

    /**
     * Get the specific aggregating state.
     *
     * @param stateDeclaration of this state.
     * @return the aggregating state corresponds to the state declaration.
     * @throws RuntimeException if the state is not available.
     */
    <IN, ACC, OUT> AggregatingState<IN, OUT> getState(
            AggregatingStateDeclaration<IN, ACC, OUT> stateDeclaration) throws Exception;

    /**
     * Get the optional of the specific broadcast state.
     *
     * @param stateDeclaration of this state.
     * @return the broadcast state corresponds to the state declaration, this may be empty.
     */
    <K, V> Optional<BroadcastState<K, V>> getStateOptional(
            BroadcastStateDeclaration<K, V> stateDeclaration) throws Exception;

    /**
     * Get the specific broadcast state.
     *
     * @param stateDeclaration of this state.
     * @return the broadcast state corresponds to the state declaration.
     * @throws RuntimeException if the state is not available.
     */
    <K, V> BroadcastState<K, V> getState(BroadcastStateDeclaration<K, V> stateDeclaration)
            throws Exception;
}

PartitionedContext에 접근할 수 있는 어디서든 context#getStateManager()를 사용해 StateManager를 얻을 수 있어요. 그런 다음 getState 또는 getStateOptional 메서드를 사용해 선언한 상태를 얻고, 읽고 갱신해요.

상태 선언과 접근의 합법성 (The Legitimacy of State Declaration and Access)

ProcessFunction에서 모든 상태 선언과 접근이 합법적인 것은 아니며, 입력 스트림의 구체적인 타입에 따라 달라져요.

OneInputStreamProcessFunction과 TwoOutputStreamProcessFunction의 경우 상태 선언과 접근의 합법성은 다음 표에 나열돼요:

두 입력의 경우는 좀 더 복잡해져요. 서로 다른 입력이 서로 다른 타입의 스트림에서 올 수 있기 때문에, 각 입력 에지의 데이터 처리 함수에 대한 상태 접근의 합법성도 달라져요. 구체적인 규칙은 다음 표에 나와 있어요:

(K, NK, G, B는 각각 Keyed, Non-Keyed, Global, Broadcast Stream의 약자예요)

Flink는 상태 선언의 합법성을 미리 검사해요. usesStates 메서드에서 불법적인 상태가 선언되면, 작업 컴파일 시점에 예외가 던져져요.

상태 접근 메서드 (State Access Methods)

datastream v2는 state v2 API를 사용한다는 점에 유의하세요. 상태 접근 메서드에 대한 자세한 내용은 Using Keyed State V2를 참고하세요.

더 알아보기 (Learn more)

• DataStream API (V2) 개요
• Building Blocks
• Working with State