Windows

Windows

Windows 는 무한 스트림 처리의 핵심입니다. Windows 는 스트림을 유한 크기의 "버킷(buckets)" 으로 분할하며, 그 위에 계산을 적용할 수 있습니다. 이 문서는 Flink DataStream 에서 윈도우잉이 수행되는 방식과 프로그래머가 그 제공 기능을 최대한 활용할 수 있는 방법에 초점을 맞춥니다.

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

출처: 문서

본문

Windows 는 무한 스트림 처리의 핵심입니다. Windows 는 스트림을 유한 크기의 "버킷(buckets)" 으로 분할하며, 그 위에 계산을 적용할 수 있습니다. 이 문서는 Flink DataStream 에서 윈도우잉이 수행되는 방식과 프로그래머가 그 제공 기능을 최대한 활용할 수 있는 방법에 초점을 맞춥니다.

DataStream API 가 제공하는 Window 기능을 활용하려면 사용자는 다음 세 단계를 완료해야 합니다:

1. Window 선언: 사용하려는 window 유형을 지정합니다.
2. WindowProcessFunction 정의: Window 의 수명 주기의 다양한 단계에서 실행되어야 하는 로직을 개요로 설명합니다.
3. Window 선언과 WindowProcessFunction 결합: window 선언과 WindowProcessFunction 을 단일 ProcessFunction 으로 캡슐화합니다. 그러면 DataStream API 내에서 활용할 수 있습니다.

이 섹션은 이 세 단계에 대한 포괄적인 개요를 제공하고 DataStream API 에서 Window 를 사용하는 방법을 보여주는 예제를 포함합니다.

Window 선언 (Declare Window)

사용자는 먼저 애플리케이션에 어떤 유형의 window 를 사용할지 결정해야 합니다. 현재 세 가지 내장 window 유형을 제공합니다: Time Window, Session Window, Global Window.

Time Window

Time Windows 는 시간 범위에 따라 여러 window 로 나뉘며, 데이터는 타임스탬프에 따라 해당 window 에 할당됩니다. 두 가지 유형의 Time Window 를 지원합니다: tumbling windows 와 sliding windows. 이러한 window 내의 시간 의미론은 event time 또는 processing time 으로 분류될 수 있습니다.

Time Windows 는 현재 Keyed Partition Stream 에서만 지원됩니다.

Tumbling Window

Tumbling Window 는 각 요소를 지정된 window size 의 window 에 할당합니다. Tumbling windows 는 고정된 크기를 가지며 겹치지 않습니다. 예를 들어 크기가 5분인 tumbling window 를 지정하면 현재 window 가 평가되고 5분마다 새 window 가 시작됩니다.

다음 코드 조각은 tumbling windows 를 사용하는 방법을 보여줍니다:

// create a tumbling window strategy with a window size of 60 seconds
WindowStrategy windowStrategy = WindowStrategy.tumbling(Duration.ofSeconds(60), WindowStrategy.EVENT_TIME);

시간 간격은 Duration.ofMillis(x), Duration.ofSeconds(x), Duration.ofMinutes(x) 등 중 하나를 사용해 지정할 수 있습니다.

Sliding Window

Sliding Window 는 요소를 고정 길이의 window 에 할당합니다. tumbling window 와 유사하게 window 의 크기는 window size 매개변수로 구성됩니다. 추가 window slide 매개변수는 sliding window 가 얼마나 자주 시작되는지 제어합니다. 따라서 slide 가 window size 보다 작으면 sliding windows 는 겹칠 수 있습니다. 이 경우 요소는 여러 window 에 할당됩니다.

예를 들어 10분 크기이고 5분씩 슬라이드하는 window 를 가질 수 있습니다. 이를 통해 5분마다 지난 10분 동안 도착한 이벤트를 포함하는 window 를 얻습니다.

다음 코드 조각은 sliding windows 를 사용하는 방법을 보여줍니다:

// create a sliding window strategy with a window size of 60 seconds and a slide of 30 seconds
WindowStrategy windowStrategy = WindowStrategy.sliding(Duration.ofSeconds(60), Duration.ofSeconds(30), WindowStrategy.PROCESSING_TIME);

시간 간격은 Duration.ofMillis(x), Duration.ofSeconds(x), Duration.ofMinutes(x) 등 중 하나를 사용해 지정할 수 있습니다.

허용 지연 (Allowed Lateness)

event-time 윈도우잉으로 작업할 때 요소가 늦게 도착할 수 있습니다. 즉, 요소가 속한 window 의 끝 타임스탬프를 Flink 가 event time 의 진행을 추적하는 데 사용하는 event time watermark 가 이미 지나갔을 수 있습니다. Flink 가 event time 을 다루는 방법에 대한 더 철저한 논의는 event time 을 참고하세요.

기본적으로 event time watermark 가 window 의 끝을 지나면 늦은 요소는 버려집니다. 그러나 Flink 는 window operator 에 대해 최대 allowed lateness 를 지정할 수 있게 합니다. Allowed lateness 는 요소가 버려지기 전에 얼마나 늦을 수 있는지를 지정하며 기본값은 0입니다.

event time watermark 가 window 끝을 지난 후 도착했지만 window 끝 + allowed lateness 를 지나기 전에 도착한 요소는 여전히 window 에 추가됩니다. 늦었지만 버려지지 않은 요소는 window 가 다시 발화(fire)되게 하며, 늦고 버려진 요소는 WindowProcessFunction#onLateRecord 에서 처리됩니다.

이를 위해 Flink 는 allowed lateness 가 만료될 때까지 window 의 상태를 유지합니다. 그렇게 되면 Flink 는 Window Lifecycle 섹션 에서 설명된 대로 window 를 제거하고 상태를 삭제합니다.

기본적으로 allowed lateness 는 0 으로 설정됩니다. 즉, watermark 뒤에 도착하는 요소는 버려집니다.

다음과 같이 allowed lateness 를 지정할 수 있습니다:

// create a sliding window strategy with a window size of 60 seconds and a slide of 30 seconds and an allowed lateness of 10 seconds
WindowStrategy windowStrategy = WindowStrategy.sliding(Duration.ofSeconds(60), Duration.ofSeconds(30), WindowStrategy.PROCESSING_TIME, Duration.ofSeconds(10));

Session Window

session windows 는 활동의 세션(session) 으로 요소를 그룹화합니다. Session windows 는 tumbling windows 및 sliding windows 와 달리 겹치지 않으며 고정된 시작과 끝 시간이 없습니다. 대신 session window 는 일정 기간 동안 요소를 받지 못하면 닫힙니다. 즉, 비활동의 간격(gap) 이 발생했을 때입니다. session window 는 session gap 으로 구성할 수 있습니다. 이 기간이 만료되면 현재 세션이 닫히고 후속 요소가 새 세션 window 에 할당됩니다.

다음 코드 조각은 session windows 를 선언하는 방법을 보여줍니다:

// create a session window strategy with a session gap of 60 seconds
WindowStrategy windowStrategy = WindowStrategy.session(Duration.ofSeconds(60), WindowStrategy.EVENT_TIME);

Session Windows 는 Global Stream 과 Keyed Partition Stream 에서만 지원됩니다.

Global Window

Global Window 는 모든 요소가 단일 통합 Window 에 할당되는 시나리오를 말합니다. 이 윈도우잉 방식은 모든 입력이 종료된 후 한 번 트리거될 수 있으므로 유한(bounded) 스트림 시나리오에서 특히 유용합니다. Global Stream, Keyed Partition Stream, Non-Keyed Partition Stream 과 호환됩니다.

다음 코드 조각은 global window 를 선언하는 방법을 보여줍니다:

// create a global window strategy
WindowStrategy windowStrategy = WindowStrategy.global();

WindowProcessFunction 정의

Window 수명 주기 (Window Lifecycle)

요컨대 window 는 이 window 에 속해야 하는 첫 번째 요소가 도착하는 즉시 생성 되고, 시간(event 또는 processing time) 이 끝 타임스탬프 + 사용자 지정 allowed lateness( Allowed Lateness 참고) 를 지나면 window 는 완전히 제거 됩니다. 예를 들어 5분마다 겹치지 않는(tumbling) window 를 만들고 allowed lateness 가 1분인 event-time 기반 윈도우잉의 경우, 12:00 에서 12:05 사이의 간격에 대한 새 window 는 이 간격에 속하는 타임스탬프를 가진 첫 요소가 도착할 때 Flink 에 의해 생성되고, event time watermark 가 12:06 타임스탬프를 지날 때 제거됩니다.

Flink 는 window 수명 주기 내의 필수 동작을 WindowProcessFunction 의 메서드로 추상화합니다. 사용자는 자신의 특정 윈도우 계산 로직을 정의하기 위해 자신의 WindowProcessFunction 을 구현해야 합니다.

WindowProcessFunction 소개

WindowProcessFunction 은 window 의 데이터를 처리하기 위해 사용자가 구현해야 하는 주요 컴포넌트입니다. window 를 선언한 후 사용자는 window 의 수명 주기의 다양한 단계에 대한 연산 로직, 즉 WindowProcessFunction 을 정의해야 합니다.

다음은 OneInputWindowStreamProcessFunction 의 인터페이스입니다:

/**
 * A type of {@link WindowProcessFunction} for one-input window processing.
 *
 * @param <IN> The type of the input value.
 * @param <OUT> The type of the output value.
 */
@Experimental
public interface OneInputWindowStreamProcessFunction<IN, OUT> extends WindowProcessFunction {

    /**
     * This method will be invoked when a record is received. Its default behaviors to store data in
     * built-in window state by {@link OneInputWindowContext#putRecord}. If the user overrides this
     * method, they have to take care of the input data themselves.
     */
    default void onRecord(
            IN record,
            Collector<OUT> output,
            PartitionedContext<OUT> ctx,
            OneInputWindowContext<IN> windowContext)
            throws Exception {
        windowContext.putRecord(record);
    }

    /**
     * This method will be invoked when the Window is triggered, you can obtain all the input
     * records in the Window by {@link OneInputWindowContext#getAllRecords()}.
     */
    void onTrigger(
            Collector<OUT> output,
            PartitionedContext<OUT> ctx,
            OneInputWindowContext<IN> windowContext)
            throws Exception;

    /**
     * Callback when a window is about to be cleaned up. It is the time to deletes any state in the
     * {@code windowContext} when the Window expires (the event time or processing time passes its
     * {@code maxTimestamp} + {@code allowedLateness}).
     */
    default void onClear(
            Collector<OUT> output,
            PartitionedContext<OUT> ctx,
            OneInputWindowContext<IN> windowContext)
            throws Exception {}

    /** This method will be invoked when a record is received after the window has been cleaned. */
    default void onLateRecord(IN record, Collector<OUT> output, PartitionedContext<OUT> ctx)
            throws Exception {}
}

WindowProcessFunction 에는 네 가지 핵심 메서드가 있습니다. 이러한 메서드의 이름과 의미는 다음과 같습니다:

• onRecord: onRecord 는 window 가 레코드를 받았음을 나타냅니다.
• onTrigger: onTrigger 는 window 가 트리거되었음을 나타냅니다.
• onClear: onClear 는 window 가 정리되었음을 나타냅니다.
• onLateRecord: onLateRecord 는 window 가 정리된 후 레코드를 받았음을 나타냅니다.

이러한 메서드에 관해 고려해야 할 몇 가지 중요한 사항이 있습니다:

• Window 는 여러 번 트리거될 수 있습니다. 따라서 onRecord 는 onTrigger 이후에 호출될 수 있습니다.
• GlobalWindow 는 데이터 스트림이 끝날 때 정리되는 반면, time/session windows 는 window 경계에 도달하고 allowedLateness( Allowed Lateness 참고) 가 경과한 후 정리됩니다.
• onLateRecord 메서드는 window 가 정리되었으므로 window 상태에 접근할 수 없습니다.

사용자는 자신의 계산 로직을 완료하기 위해 WindowProcessFunction 을 구현해야 합니다. 이 단계에서 Window State 를 사용해 Window 관련 데이터를 저장하고 Window 내장 상태 접근 을 통해 Window 데이터를 저장하는 데 사용되는 내장 상태에 접근할 수 있습니다.

Window State 선언과 사용

window 에는 두 가지 유형의 상태가 있습니다: partitioned state 와 window state.

• Partitioned State: 파티션 관련 상태를 partitioned state 라고 합니다. NonKeyedStream 의 경우 이 상태는 특정 task 간에 공유됩니다. KeyedStream 의 경우 이 상태는 같은 키를 가진 데이터 간에 공유됩니다. 사용자는 ProcessFunction#usesStates 를 통해 partitioned state 를 선언하고 PartitionedContext#getStateManager 를 통해 partitioned state 를 사용할 수 있습니다. WindowProcessFunction#onClear 에서 더 이상 필요하지 않은 partitioned state 의 데이터를 정리하는 것은 사용자의 책임입니다.

• Window State: window 관련 상태를 window state 라고 합니다. Window state 는 특정 window 에 바인딩됩니다. 예를 들어 10:00-11:00 window 에서 같은 키에 대해 선언되고 사용된 window state 는 11:00-12:00 window 의 것과 다릅니다. 사용자는 WindowProcessFunction#usesWindowStates 를 통해 window state 를 선언하고 WindowContext#getWindowState 를 통해 window state 를 사용할 수 있습니다. 모든 window state 는 사용자가 WindowProcessFunction#onClear 에서 수동으로 지우는지 여부와 관계없이 결국 프레임워크에 의해 정리됩니다.

Window 내장 상태 접근

입력 데이터를 저장하기 위해 각 window 에 대해 내장 window 상태를 제공합니다. 사용자는 WindowContext#putRecord 와 WindowContext#getAllRecords 를 통해 이에 접근할 수 있습니다. 이 상태는 window 가 정리될 때 지워집니다.

기본적으로 WindowProcessFunction#onRecord 는 WindowContext#putRecord 를 통해 받은 데이터를 window 의 내장 상태에 저장하며, 사용자는 window 가 트리거될 때 WindowContext#getAllRecords 를 사용해 Window 내의 모든 데이터를 검색할 수 있습니다.

따라서 WindowProcessFunction#onRecord 를 재정의할 때 사용자는 입력 데이터를 내장 상태에 써야 하는지 고려해야 합니다. 전형적인 예로 사전 집계(pre-aggregation) 를 수행하려는 경우, Window 상태를 선언하고 WindowProcessFunction#onRecord 에서 집계를 수행하며 집계된 window 상태를 업데이트하고 WindowProcessFunction#onTrigger 에서 최종 결과를 출력할 수 있습니다. 따라서 모든 데이터를 캐싱하는 불필요한 비용이 제거됩니다.

ProcessFunction 구축

Window 를 선언하고 WindowProcessFunction 을 정의한 후 사용자는 BuiltinFuncs.window 메서드를 활용해 이 두 컴포넌트를 데이터 처리 스트림에 통합될 수 있는 ProcessFunction 으로 캡슐화해야 합니다. 예는 다음과 같습니다:

KeyedPartitionStream stream = ...;
OneInputStreamProcessFunction wrappedWindowProcessFunction = BuiltinFuncs.window(windowStrategy, new CustomWindowProcessFunction());
stream.process(wrappedWindowProcessFunction)
      .process(...);

이렇게 하면 Flink 가 window 데이터를 캐싱하는 데 필요한 상태와 타이머를 자동으로 관리하므로 사용자는 이러한 세부 사항을 직접 처리해야 하는 부담에서 벗어납니다.

예제: 매시간 각 제품의 판매량 계산

다음은 Window 를 사용해 각 시간에 각 제품의 판매량을 계산하는 방법의 예입니다.

다음 예제에서 먼저 event time 확장 을 사용해 orderSource 에서 event time 을 추출한 다음, 1시간 tumbling window 를 선언하고 window 가 트리거될 때 각 제품의 판매 수량을 계산합니다.

public class CountProductSalesEveryHour {

    public static class Order {
        public long orderId;
        public long productId;
        public long salesQuantity;
        public long orderTime;
    }

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getInstance();

        // create order source stream
        NonKeyedPartitionStream<Order> orderSource = ...;

        // extract and propagate event time from order
        NonKeyedPartitionStream<Order> orderStream = orderSource.process(
                EventTimeExtension
                        .<Order>newWatermarkGeneratorBuilder(order -> order.orderTime)
                        .periodicWatermark(Duration.ofMillis(200))
                        .buildAsProcessFunction()
        );

        NonKeyedPartitionStream<Tuple2<Long, Long>> productSalesQuantityStream = orderStream
                // key by productId
                .keyBy(order -> order.productId)
                .process(BuiltinFuncs.window(
                                // declare tumbling window with window size 10 seconds
                                WindowStrategy.tumbling(
                                        Duration.ofHours(1),
                                        WindowStrategy.EVENT_TIME),
                                // define window process function to calculate total sales quantity per product per window
                                new CountSalesQuantity()
                        )
                );

        // print result
        productSalesQuantityStream.toSink(new WrappedSink<>(new PrintSink<>()));

        env.execute("CountSalesQuantifyOfEachProductEveryHour");
    }

    public static class CountSalesQuantity implements OneInputWindowStreamProcessFunction<Order, Tuple2<Long, Long>>  {

        @Override
        public void onTrigger(
                Collector<Tuple2<Long, Long>> output,
                PartitionedContext<Tuple2<Long, Long>> ctx,
                OneInputWindowContext<Order> windowContext) throws Exception {
            // get current productId
            long productId = ctx.getStateManager().getCurrentKey();
            // calculate total sales quantity
            long totalSalesQuantity = 0;
            for (Order order : windowContext.getAllRecords()) {
                totalSalesQuantity += order.salesQuantity;
            }
            // emit result
            output.collect(Tuple2.of(productId, totalSalesQuantity));
        }
    }
}

이 예제의 시나리오에서 사용자는 Window 에서 사전 집계를 수행해 Window 계산의 비용을 줄일 수 있습니다. CountSalesQuantity WindowProcessFunction 을 입력 데이터가 도착할 때 사전 집계를 수행할 수 있는 CountSalesQuantityWithPreAggregation WindowProcessFunction 으로 다시 작성했습니다.

public static class CountSalesQuantityWithPreAggregation implements OneInputWindowStreamProcessFunction<Order, Tuple2<Long, Long>>  {

    private final ValueStateDeclaration<Long> salesQuantityStateDeclaration =
            StateDeclarations.valueState("totalSalesQuantity", TypeDescriptors.LONG);

    @Override
    public Set<StateDeclaration> useWindowStates() {
        return Set.of(salesQuantityStateDeclaration);
    }

    @Override
    public void onRecord(
            Order record,
            Collector<Tuple2<Long, Long>> output,
            PartitionedContext<Tuple2<Long, Long>> ctx,
            OneInputWindowContext<Order> windowContext) throws Exception {
        // get sales quantity from state
        ValueState<Long> salesQuantityState = windowContext.getWindowState(salesQuantityStateDeclaration).get();
        long salesQuantity = 0;
        if (salesQuantityState.value() != null) {
            salesQuantity = salesQuantityState.value();
        }

        // update sales quantity in state
        salesQuantity += record.salesQuantity;
        salesQuantityState.update(salesQuantity);
    }

    @Override
    public void onTrigger(
            Collector<Tuple2<Long, Long>> output,
            PartitionedContext<Tuple2<Long, Long>> ctx,
            OneInputWindowContext<Order> windowContext) throws Exception {
        // get current productId
        long productId = ctx.getStateManager().getCurrentKey();
        // get sales quantity from state
        ValueState<Long> salesQuantityState = windowContext.getWindowState(salesQuantityStateDeclaration).get();
        long salesQuantity = salesQuantityState.value() == null ? 0 : salesQuantityState.value();
        // emit result
        output.collect(Tuple2.of(productId, salesQuantity));
    }
}

CountSalesQuantityWithPreAggregation 함수에서 먼저 window 내의 각 제품에 대한 총 판매 수량을 저장할 ValueState 를 선언합니다. 입력 데이터가 도착하면 이 상태를 그에 맞게 업데이트하고, window 가 트리거되면 최종 결과를 출력합니다. 이 접근 방식은 모든 입력 데이터를 window 에 저장하는 것을 피할 수 있으며, window 내의 각 제품에 대해 ValueState 하나만 유지하면 됩니다.

더 알아보기 (Learn more)

• DataStream API V2 개요
• 시간 처리 (Time Processing)
• Event Timer Service