Processing Timer Service

Processing Timer Service (처리 시간 타이머 서비스)

참고: DataStream API V2는 기존 DataStream API를 점차 대체하기 위한 새로운 API 집합입니다. 현재 실험 단계이며 프로덕션에서 완전히 사용할 수는 없습니다.

프로세스 타이머 서비스는 Flink가 제공하는 DataStream API의 기본 프리미티브(primitive)입니다. 이를 통해 사용자는 특정 처리 시간(processing time) 시점에 계산을 수행하기 위한 타이머를 등록할 수 있습니다.

출처: 문서

본문

프로세스 타이머 서비스는 Flink가 제공하는 DataStream API의 기본 프리미티브입니다. 이를 통해 사용자는 특정 처리 시간 시점에 계산을 수행하기 위한 타이머를 등록할 수 있습니다.

처리 시간에 대한 포괄적인 설명은 Notions of Time: Event Time and Processing Time 섹션을 참고하세요.

이 섹션에서는 Flink DataStream API 내에서 처리 타이머 서비스를 활용하는 방법을 소개합니다.

ProcessingTimerManager

ProcessingTimerManager는 처리 타이머 서비스를 활용하기 위한 핵심 컴포넌트입니다. ProcessFunction 내에서 PartitionedContext#getProcessingTimeManager를 호출하여 얻을 수 있습니다. 아래는 ProcessingTimerManager의 인터페이스입니다:

@Experimental
public interface ProcessingTimeManager {
    /**
     * Register a processing timer for this process function. `onProcessingTimer` method of this
     * function will be invoked as callback if the timer expires.
     *
     * @param timestamp to trigger timer callback.
     */
    void registerTimer(long timestamp);

    /**
     * Deletes the processing-time timer with the given trigger timestamp. This method has only an
     * effect if such a timer was previously registered and did not already expire.
     *
     * @param timestamp indicates the timestamp of the timer to delete.
     */
    void deleteTimer(long timestamp);

    /**
     * Get the current processing time.
     *
     * @return current processing time.
     */
    long currentTime();

}

위에서 보듯이 ProcessingTimerManager는 registerTimer, deleteTimer, currentTime의 세 가지 메서드를 가집니다. registerTimer의 목표 시간에 도달하면 Flink가 ProcessFunction#onProcessingTimer를 호출하며, 사용자는 ProcessFunction#onProcessingTimer에 자신의 계산 로직을 작성해야 합니다.

주의할 점:

같은 목표 시간을 가진 타이머에 대해 Flink는 하나의 타이머만 유지합니다. 즉, 같은 목표 시간의 여러 타이머가 등록되어도 ProcessFunction#onProcessingTimer는 한 번만 호출됩니다.
ProcessingTimerManager는 Keyed Partitioned Stream에서만 사용할 수 있습니다.

예제 (Example)

다음은 처리 타이머 서비스를 사용하는 예제입니다:

public class CustomProcessFunction implements OneInputStreamProcessFunction<String, String> {

    @Override
    public void processRecord(
            String record,
            Collector<String> output,
            PartitionedContext<String> ctx) throws Exception {
        // do some calculation as needed

        // register a timer with a target time of one minute after the current time
        long currentTime = ctx.getProcessingTimeManager().currentTime();
        ctx.getProcessingTimeManager().registerTimer(currentTime + Duration.ofMinutes(1L).toMillis());
    }

    @Override
    public void onProcessingTimer(
            long timestamp,
            Collector<String> output,
            PartitionedContext<String> ctx) {
        // do some calculation and output result as needed
    }

}

Processing Timer Service