[Concept] 4. Flink에서의 Time Semantics과 Watermark

1. Overview

스트림 처리에서 시간(Time)은 매우 중요한 개념이다. 어떤 시간을 기준으로 연산을 수행하느냐에 따라 결과가 완전히 달라진다. Apache Flink는 이러한 문제를 해결하기 위해 시간을 Event Time, Ingestion Time, Processing Time으로 구분한다.

일반적으로 Event Time을 사용하면 실제 이벤트가 발생한 시점을 기준으로 데이터를 처리한다. 하지만 Event Time을 사용할 때는 이벤트가 순서대로 도착하지 않는 문제가 발생할 수 있다. 이 문제를 해결하기 위해 Flink는 Watermark를 사용한다. Watermark는 늦게 도착하는 데이터를 얼마나 기다릴지를 결정하는 기준이며, 보통 일정한 주기로 생성된다.

2. Time Semantics

2-1. Event Time

Event Time은 실제로 이벤트가 발생한 그 순간의 시간을 의미한다. 예를 들어 사용자가 버튼을 클릭하거나, 센서가 값을 측정한 시점이 이에 해당한다.

Event Time을 사용하려면 이벤트에 포함된 타임스탬프를 Flink에 전달해야 한다. 일반적으로 타임스탬프는 이벤트 자체에 포함되어 있으며, Flink에서는 해당 필드를 추출해 사용한다.

KafkaSource<Event> source =
    KafkaSource.<Event>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics("events")
        .setGroupId("flink-consumer")
        .setValueOnlyDeserializer(new EventDeserializer())
        .build();

DataStream<Event> events = env.fromSource(
    source,
    WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((event, timestamp) -> event.getTimestamp()),
    "Kafka Source"
);

이 방식은 네트워크 지연과 무관하게 정확한 시간 기준 연산이 가능하며, 과거 데이터를 재처리하더라도 동일한 결과를 보장한다는 점이 장점이다. 반면 이벤트 순서가 보장되지 않는다는 문제가 발생할 수 있으며, 이는 Watermark를 통해 해결한다.

2-2. Processing Time

Processing Time은 연산자가 이벤트를 처리하는 시점의 시간이다. 이벤트가 TaskManager에 도착해 처리되는 순간의 시스템 시간을 기준으로 한다.

가장 큰 장점은 단순함이다. 타임스탬프 추출이나 워터마크 설정이 필요 없으며, 시스템 시간(System.currentTimeMillis)을 그대로 사용한다.

이 방식은 레이턴시가 매우 낮아 실시간 모니터링과 같은 용도에 적합하다. 다만 동일한 데이터를 다시 처리하더라도 처리 시점에 따라 결과가 달라질 수 있다는 문제가 존재한다.

2-3. Ingestion Time

Ingestion Time은 이벤트가 Flink의 Source 연산자에 도착한 시점의 시간을 의미한다. Event Time과 Processing Time의 중간 개념으로, Source가 자동으로 타임스탬프를 할당한다.

이 방식은 이벤트에서 타임스탬프를 직접 추출할 필요가 없기 때문에 설정이 단순하다. 그러나 실제 이벤트 발생 시점과는 차이가 발생할 수 있으며, 네트워크 지연을 반영하지 못한다는 한계가 있다.

또한 Flink 1.12 이후에는 Ingestion Time 설정이 deprecated 되었으며, 현재는 Event Time을 기본 시간 기준으로 사용하는 것이 권장된다.

env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

이러한 이유로 Ingestion Time은 실무에서 자주 사용되지는 않는다.

3. Watermark

3-1. Out-of-Order 문제와 Watermark

Event Time을 사용할 때는 이벤트가 발생한 순서와 Flink에 도착한 순서가 다를 수 있다. 이를 Out-of-Order 문제라고 하며, Flink는 이를 완화하기 위해 허용 지연 시간(allowed lateness) 개념을 사용한다. 즉, 일정 시간 동안은 늦게 도착한 이벤트도 정상적으로 처리될 수 있도록 기다린다. 이 지연 시간을 기준으로 Flink는 Watermark를 생성한다.

DataStream<Event> eventsWithWatermark = env
    .fromSource(
        source,
        WatermarkStrategy
            // 최대 10초까지 순서가 뒤바뀔 수 있다고 가정
            .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner((event, timestamp) -> event.getTimestamp()),
        "Kafka Source"
    )
    .keyBy(event -> event.getSensorId())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.seconds(30))   // 30초까지 늦은 데이터 허용
    .sideOutputLateData(lateDataTag)     // 더 늦은 건 side output으로 분리
    .sum("temperature")
    .print();

허용 지연 시간을 초과한 데이터는 Side Output으로 분리해 별도로 처리할 수 있다.

DataStream<Event> lateEvents =
    eventsWithWatermark.getSideOutput(lateDataTag);

예를 들어 가장 늦게 도착한 이벤트의 타임스탬프가 10시 30초이고, 허용 지연이 5초라면 Watermark는 10시 25초로 설정된다.

3-2. Watermark 생성 방식

Watermark를 만드는 방법은 크게 두 가지가 있다.

첫 번째는 Periodic Watermark로, 일정한 시간 간격마다 자동으로 Watermark를 만들어내는 방식이다. 대부분의 상황에서 이 방법이 사용된다.

두 번째는 Punctuated Watermark로, 특정한 이벤트가 들어올 때만 Watermark를 만드는 방식이다. 예를 들어 로그의 끝을 알리는 특별한 이벤트가 있을 때 그 시점에 Watermark를 생성하는 식이다.

public class PunctuatedWatermarkGenerator
        implements WatermarkGenerator<Event> {

   @Override
   public void onEvent(
           Event event,
           long eventTimestamp,
           WatermarkOutput output) {

       if (event.hasWatermarkMarker()) {
           output.emitWatermark(
               new Watermark(event.getTimestamp())
           );
       }
   }

   @Override
   public void onPeriodicEmit(WatermarkOutput output) {
       // Punctuated Watermark는 주기적으로 생성하지 않는다
   }
}

Punctuated Watermark는 이렇게 특별한 이벤트가 있어야 동작하기 때문에 실제로는 잘 쓰이지 않는다. 대부분의 경우에는 Periodic Watermark만으로 충분하다.