Process Function

Process Function (프로세스 함수)

ProcessFunction은 스트림 처리의 저수준 기본 요소(이벤트, 상태, 타이머)에 접근할 수 있게 해주는 연산이에요. 특히 keyed state와 타이머가 필요한 커스텀 처리 로직을 구현할 때 유용해요.

출처: Process Function

본문

The ProcessFunction

ProcessFunction은 모든 (비순환) 스트리밍 애플리케이션의 기본 구성 요소에 접근할 수 있게 해주는 저수준 스트림 처리 연산이에요:

  • 이벤트 (스트림 요소)
  • 상태 (내결함성, 일관성, keyed stream에서만)
  • 타이머 (이벤트 시간과 처리 시간, keyed stream에서만)

ProcessFunction은 keyed state와 타이머에 접근할 수 있는 FlatMapFunction으로 생각할 수 있어요. 입력 스트림에서 수신된 각 이벤트마다 호출되어 이벤트를 처리해요.

내결함성 상태를 위해 ProcessFunction은 Flink의 [keyed state]({{< ref "docs/dev/datastream/fault-tolerance/state" >}})에 접근할 수 있게 해주고, 다른 상태 기반 함수가 keyed state에 접근하는 것과 비슷한 방식으로 RuntimeContext를 통해 접근해요.

타이머를 사용하면 애플리케이션이 처리 시간과 [이벤트 시간]({{< ref "docs/concepts/time" >}})의 변화에 반응할 수 있어요. processElement(...) 함수에 대한 매 호출은 요소의 이벤트 시간 타임스탬프와 TimerService에 접근할 수 있는 Context 객체를 받아요. TimerService는 미래의 이벤트/처리 시간 인스턴스에 대한 콜백을 등록하는 데 사용될 수 있어요. 이벤트 시간 타이머의 경우 현재 워터마크가 타이머의 타임스탬프까지 진행되면 onTimer(...) 메서드가 호출되고, 처리 시간 타이머의 경우 벽시계 시간이 지정된 시간에 도달하면 onTimer(...)가 호출돼요. 그 호출 동안 모든 상태는 타이머가 생성된 키로 다시 범위가 지정되어, 타이머가 keyed state를 조작할 수 있게 해줘요.

{{< hint info >}} keyed state와 타이머에 접근하려면 ProcessFunction을 keyed stream에 적용해야 해요: {{< /hint >}}

stream.keyBy(...).process(new MyProcessFunction());

Low-level Joins (저수준 조인)

두 입력에 대한 저수준 연산을 구현하기 위해 애플리케이션은 CoProcessFunction 또는 KeyedCoProcessFunction을 사용할 수 있어요. 이 함수는 두 개의 서로 다른 입력에 바인딩되며, 두 입력의 레코드에 대해 processElement1(...)와 processElement2(...) 호출을 각각 받아요.

저수준 조인을 구현하는 일반적인 패턴은 다음과 같아요:

  • 한 입력(또는 둘 다)에 대한 상태 객체를 만들어요
  • 해당 입력에서 요소를 받을 때 상태를 갱신해요
  • 다른 입력에서 요소를 받으면 상태를 조회하고 조인 결과를 생성해요

예를 들어 고객 데이터를 금융 거래에 조인하면서 고객 데이터에 대한 상태를 유지하는 경우를 생각해볼 수 있어요. 순서가 어긋난 이벤트 앞에서도 완전하고 결정적인 조인을 원한다면, 고객 데이터 스트림의 워터마크가 해당 거래의 시간을 지나면 타이머를 사용해 거래에 대한 조인을 평가하고 방출할 수 있어요.

Example (예제)

다음 예제에서 KeyedProcessFunction은 키별 카운트를 유지하고, 해당 키에 대한 업데이트 없이 (이벤트 시간 기준) 1분이 지나면 키/카운트 쌍을 방출해요:

  • 카운트, 키, 마지막 수정 타임스탬프는 키로 암시적으로 범위가 지정된 ValueState에 저장돼요.
  • 각 레코드에 대해 KeyedProcessFunction은 카운터를 증가시키고 마지막 수정 타임스탬프를 설정해요
  • 함수는 또한 (이벤트 시간 기준) 1분 후의 콜백을 스케줄링해요
  • 각 콜백에서 콜백의 이벤트 시간 타임스탬프를 저장된 카운트의 마지막 수정 시간과 비교하고, 일치하면 (즉 그 1분 동안 추가 업데이트가 없었다면) 키/카운트를 방출해요

{{< hint info >}} 이 간단한 예제는 세션 윈도우로 구현할 수 있었어요. 여기서는 기본 패턴을 설명하기 위해 KeyedProcessFunction을 사용해요. {{< /hint >}}

import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction.Context;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction.OnTimerContext;
import org.apache.flink.util.Collector;


// the source data stream
DataStream<Tuple2<String, String>> stream = ...;

// apply the process function onto a keyed stream
DataStream<Tuple2<String, Long>> result = stream
    .keyBy(value -> value.f0)
    .process(new CountWithTimeoutFunction());

/**
 * The data type stored in the state
 */
public class CountWithTimestamp {

    public String key;
    public long count;
    public long lastModified;
}

/**
 * The implementation of the ProcessFunction that maintains the count and timeouts
 */
public class CountWithTimeoutFunction 
        extends KeyedProcessFunction<Tuple, Tuple2<String, String>, Tuple2<String, Long>> {

    /** The state that is maintained by this process function */
    private ValueState<CountWithTimestamp> state;

    @Override
    public void open(OpenContext openContext) throws Exception {
        state = getRuntimeContext().getState(new ValueStateDescriptor<>("myState", CountWithTimestamp.class));
    }

    @Override
    public void processElement(
            Tuple2<String, String> value, 
            Context ctx, 
            Collector<Tuple2<String, Long>> out) throws Exception {

        // retrieve the current count
        CountWithTimestamp current = state.value();
        if (current == null) {
            current = new CountWithTimestamp();
            current.key = value.f0;
        }

        // update the state's count
        current.count++;

        // set the state's timestamp to the record's assigned event time timestamp
        current.lastModified = ctx.timestamp();

        // write the state back
        state.update(current);

        // schedule the next timer 60 seconds from the current event time
        ctx.timerService().registerEventTimeTimer(current.lastModified + 60000);
    }

    @Override
    public void onTimer(
            long timestamp, 
            OnTimerContext ctx, 
            Collector<Tuple2<String, Long>> out) throws Exception {

        // get the state for the key that scheduled the timer
        CountWithTimestamp result = state.value();

        // check if this is an outdated timer or the latest timer
        if (timestamp == result.lastModified + 60000) {
            // emit the state on timeout
            out.collect(new Tuple2<String, Long>(result.key, result.count));
        }
    }
}

{{< hint warning >}} Flink 1.4.0 이전에는 처리 시간 타이머에서 호출될 때 ProcessFunction.onTimer() 메서드가 현재 처리 시간을 이벤트 시간 타임스탬프로 설정했어요. 이 동작은 매우 미묘해서 사용자가 알아차리지 못할 수 있어요. 처리 시간 타임스탬프는 비결정적이고 워터마크와 정렬되지 않기 때문에 해롭답니다. 게다가 이 잘못된 타임스탬프에 의존하는 사용자 구현 로직은 의도치 않게 오류가 나기 쉽습니다. 그래서 이 문제를 고치기로 했어요. 1.4.0으로 업그레이드하면 이 잘못된 이벤트 시간 타임스탬프를 사용하는 Flink 잡은 실패하며, 사용자는 잡을 올바른 로직에 맞게 조정해야 합니다. {{< /hint >}}

The KeyedProcessFunction

KeyedProcessFunction은 ProcessFunction의 확장으로, onTimer(...) 메서드에서 타이머의 키에 접근할 수 있게 해줘요.

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception {
    K key = ctx.getCurrentKey();
    // ...
}


def on_timer(self, timestamp: int, ctx: 'KeyedProcessFunction.OnTimerContext'):
    key = ctx.get_current_key()
    # ...

Timers (타이머)

두 유형의 타이머(처리 시간 및 이벤트 시간)는 내부적으로 TimerService에 의해 유지되고 실행을 위해 큐에 들어가요.

TimerService는 키와 타임스탬프별로 타이머를 중복 제거해요. 즉 키와 타임스탬프당 최대 하나의 타이머가 있어요. 같은 타임스탬프에 여러 타이머가 등록되면 onTimer() 메서드는 한 번만 호출돼요.

Flink는 onTimer()와 processElement()의 호출을 동기화해요. 따라서 사용자가 상태의 동시 수정에 대해 걱정할 필요는 없어요.

Fault Tolerance (내결함성)

타이머는 내결함성이 있으며 애플리케이션의 상태와 함께 체크포인트로 저장돼요. 장애 복구 시 또는 savepoint에서 애플리케이션을 시작할 때 타이머가 복원돼요.

{{< hint info >}} 복원 전에 실행되도록 예정된 체크포인트된 처리 시간 타이머는 즉시 실행돼요. 이는 애플리케이션이 장애에서 복구되거나 savepoint에서 시작될 때 발생할 수 있어요. {{< /hint >}}

{{< hint info >}} 타이머는 RocksDB 백엔드 / 증분 스냅샷 / 힙 기반 타이머의 조합(FLINK-10026으로 해결 예정)을 제외하면 항상 비동기로 체크포인트되어요. 타이머는 체크포인트된 상태의 일부이므로 많은 수의 타이머는 체크포인트 시간을 늘릴 수 있어요. 타이머 수를 줄이는 방법에 대한 조언은 "Timer Coalescing" 섹션을 참고하세요. {{< /hint >}}

Timer Coalescing (타이머 병합)

Flink는 키와 타임스탬프당 하나의 타이머만 유지하므로, 타이머 해상도를 줄여 타이머를 병합함으로써 타이머 수를 줄일 수 있어요.

1초 해상도의 타이머(이벤트 또는 처리 시간)에 대해 대상 시간을 전체 초로 내림할 수 있어요. 타이머는 요청된 것보다 최대 1초 일찍 실행되지만 밀리초 정확도로 늦게 실행되지는 않아요. 결과적으로 키당 초당 최대 하나의 타이머가 있어요.

long coalescedTime = ((ctx.timestamp() + timeout) / 1000) * 1000;
ctx.timerService().registerProcessingTimeTimer(coalescedTime);

coalesced_time = ((ctx.timestamp() + timeout) // 1000) * 1000
ctx.timer_service().register_processing_time_timer(coalesced_time)

이벤트 시간 타이머는 들어오는 워터마크와 함께만 실행되므로 현재 워터마크를 사용해 다음 워터마크로 이 타이머들을 스케줄링하고 병합할 수도 있어요:

long coalescedTime = ctx.timerService().currentWatermark() + 1;
ctx.timerService().registerEventTimeTimer(coalescedTime);

coalesced_time = ctx.timer_service().current_watermark() + 1
ctx.timer_service().register_event_time_timer(coalesced_time)

타이머는 다음과 같이 중지하고 제거할 수도 있어요:

처리 시간 타이머 중지:

long timestampOfTimerToStop = ...;
ctx.timerService().deleteProcessingTimeTimer(timestampOfTimerToStop);

timestamp_of_timer_to_stop = ...
ctx.timer_service().delete_processing_time_timer(timestamp_of_timer_to_stop)

이벤트 시간 타이머 중지:

long timestampOfTimerToStop = ...;
ctx.timerService().deleteEventTimeTimer(timestampOfTimerToStop);

timestamp_of_timer_to_stop = ...
ctx.timer_service().delete_event_time_timer(timestamp_of_timer_to_stop)

{{< hint info >}} 주어진 타임스탬프로 등록된 타이머가 없으면 타이머 중지는 효과가 없어요. {{< /hint >}}

더 알아보기 (Learn more)