윈도우

윈도우 (Windows)

무한 스트림을 처리할 때 가장 기본이 되는 생각이 바로 윈도우예요. 윈도우는 무한한 스트림을 유한한 크기의 '버킷'으로 잘라내고, 그 버킷 안에서 계산을 적용할 수 있게 해줘요. 이 문서는 Flink에서 윈도우잉이 어떻게 이뤄지는지, 그리고 프로그래머가 그 기능을 최대한 활용하려면 어떻게 해야 하는지에 초점을 맞춰요.

출처: Apache Flink 공식 문서 - Windows

윈도우 처리 프로그램의 구조

윈도우 처리 프로그램의 일반적인 구조를 먼저 볼게요. 첫 코드는 키 스트림(keyed stream), 두 번째는 비키 스트림(non-keyed stream)의 형태예요. 차이는 키 스트림의 keyBy(...) 호출과, 비키 스트림에서 window(...) 대신 쓰는 windowAll(...)뿐이에요.

키 스트림 (Keyed Windows) — Java

stream
    .keyBy(...)              <- 키드 vs 비키드 윈도우
    .window(...)             <- 필수: "assigner"
   [.trigger(...)]           <- 선택: "trigger" (기본 트리거 사용 시 생략)
   [.evictor(...)]           <- 선택: "evictor"
   [.allowedLateness(...)]   <- 선택: "lateness" (기본값 0)
   [.sideOutputLateData(...)]<- 선택: "output tag"
    .reduce/aggregate/apply()<- 필수: "function"
   [.getSideOutput(...)]     <- 선택: "output tag"

키 스트림 (Keyed Windows) — Python

stream
    .key_by(...)
    .window(...)                      <- 필수: "assigner"
   [.trigger(...)]                    <- 선택: "trigger"
   [.allowed_lateness(...)]           <- 선택: "lateness"
   [.side_output_late_data(...)]      <- 선택: "output tag"
    .reduce/aggregate/apply()         <- 필수: "function"
   [.get_side_output(...)]            <- 선택: "output tag"

비키 스트림 (Non-Keyed Windows)

stream
    .windowAll(...) <- 필수: "assigner"
   [.trigger(...)]  ...

대괄호([...]) 안의 명령은 선택 사항이에요. 이 구조 하나만 봐도 Flink가 원하는 대로 윈도우잉 로직을 맞춤 구성할 수 있게 해준다는 걸 알 수 있어요.

윈도우 생명주기 (Window Lifecycle)

요약하면 윈도우는 그 윈도우에 속해야 할 첫 요소가 도착하는 순간 생성되고, 시간(이벤트 시간 또는 처리 시간)이 그 종료 타임스탬프에 사용자가 지정한 allowed lateness를 더한 시점을 지나가면 완전히 제거돼요. Flink는 제거를 시간 기반 윈도우에 대해서만 보장하고, 다른 종류(예: 글로벌 윈도우)에는 보장하지 않아요. 예를 들어 5분마다 겹치지 않는(텀블링) 윈도우를 이벤트 시간으로 만들고 allowed lateness를 1분으로 줬다면, 12:00~12:05 구간의 첫 요소가 도착할 때 윈도우가 생성되고 워터마크가 12:06 타임스탬프를 지나면 제거돼요.

또한 각 윈도우에는 Trigger와 함수(ProcessWindowFunction, ReduceFunction, AggregateFunction)가 붙어요. 함수가 윈도우 내용물에 적용할 계산을 담고, Trigger가 윈도우가 함수를 적용할 준비가 됐다고 판단하는 조건을 지정해요. 트리거링 정책은 "윈도우에 요소가 4개보다 많아졌을 때" 같은 것일 수도, "워터마크가 윈도우 끝을 지나갈 때"일 수도 있어요. 트리거는 생성과 제거 사이의 아무 때나 윈도우 내용물을 비울(purge) 수도 있는데, 이때 비워지는 건 요소뿐이고 윈도우 메타데이터는 남아요. 그래서 그 윈도우에 새 데이터가 계속 추가될 수 있어요.

이 외에도 Evictor를 지정하면 트리거가 발화된 후, 함수 적용 전/후에 윈도우에서 요소를 제거할 수 있어요.

키 윈도우 vs 비키 윈도우

먼저 스트림을 키 스트림으로 만들지 결정해야 해요. 윈도우를 정의하기 전에 정해야 하죠. keyBy(...)를 쓰면 무한 스트림이 논리적 키 스트림들로 분리돼요. keyBy(...)를 호출하지 않으면 스트림은 키드 상태가 아니에요.

키 스트림의 경우 들어오는 이벤트의 어떤 속성이라도 키로 쓸 수 있어요. 키 스트림을 쓰면 윈도우 계산을 여러 태스크로 병렬 수행할 수 있는데, 각 논리적 키 스트림이 나머지와 독립적으로 처리되기 때문이에요. 같은 키를 가진 모든 요소는 같은 병렬 태스크로 보내져요. 반면 비키 스트림은 스트림이 여러 논리적 스트림으로 분리되지 않고, 모든 윈도우잉 로직이 단일 태스크(병렬도 1)에서 실행돼요.

윈도우 어사이너 (Window Assigners)

다음 단계는 윈도우 어사이너(window assigner)를 정의하는 거예요. 어사이너는 요소가 어떤 윈도우에 배정되는지를 정해요. window(...)(키 스트림) 또는 windowAll()(비키 스트림) 호출에서 원하는 WindowAssigner를 지정하면 돼요. 어사이너는 각 입력 요소를 하나 이상의 윈도우에 배정하는 역할을 해요.

Flink는 가장 흔한 용례를 위한 내장 어사이너를 제공해요: 텀블링 윈도우, 슬라이딩 윈도우, 세션 윈도우, 글로벌 윈도우. WindowAssigner 클래스를 확장해 커스텀 어사이너를 직접 구현할 수도 있어요. 글로벌 윈도우를 제외한 모든 내장 어사이너는 시간을 기준으로 요소를 윈도우에 배정하는데, 그 시간은 처리 시간일 수도 이벤트 시간일 수도 있어요.

시간 기반 윈도우는 윈도우 크기를 함께 설명하는 **시작 타임스탬프(포함)**와 **종료 타임스탬프(제외)**를 가져요. 코드에서 Flink는 시간 기반 윈도우를 다룰 때 TimeWindow를 사용하며, 시작·종료 타임스탬프를 조회하는 메서드와 주어진 윈도우에서 허용되는 가장 큰 타임스탬프를 반환하는 maxTimestamp() 메서드도 제공해요.

텀블링 윈도우 (Tumbling Windows)

텀블링 윈도우 어사이너는 각 요소를 지정된 윈도우 크기의 윈도우에 배정해요. 텀블링 윈도우는 크기가 고정되어 있고 겹치지 않아요. 예를 들어 5분 크기의 텀블링 윈도우를 지정하면, 현재 윈도우가 평가되고 5분마다 새 윈도우가 시작돼요.

DataStream<T> input = ...;

// 텀블링 이벤트-시간 윈도우
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
    .<windowed transformation>(<window function>);

// 텀블링 처리-시간 윈도우
input
    .keyBy(<key selector>)
    .window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5)))
    .<windowed transformation>(<window function>);

// -8시간 오프셋이 적용된 일 단위 텀블링 이벤트-시간 윈도우
input
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Duration.ofDays(1), Duration.ofHours(-8)))
    .<windowed transformation>(<window function>);

# 텀블링 이벤트-시간 윈도우
input \
    .key_by(<key selector>) \
    .window(TumblingEventTimeWindows.of(Duration.ofSeconds(5))) \
    .<windowed transformation>(<window function>)

# 텀블링 처리-시간 윈도우
input \
    .key_by(<key selector>) \
    .window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5))) \
    .<windowed transformation>(<window function>)

시간 간격은 Duration.ofMillis(x), Duration.ofSeconds(x), Duration.ofMinutes(x) 등으로 지정할 수 있어요.

마지막 예시처럼 텀블링 윈도우 어사이너는 윈도우 정렬을 바꾸는 선택적 offset 파라미터도 받아요. 오프셋 없이 시간 단위 텀블링 윈도우를 만들면 epoch에 정렬되어 1:00:00.000 - 1:59:59.999, 2:00:00.000 - 2:59:59.999 같은 윈도우가 나와요. 15분 오프셋을 주면 1:15:00.000 - 2:14:59.999, 2:15:00.000 - 3:14:59.999 같은 윈도우가 생기죠. 오프셋의 중요한 용례는 UTC-0이 아닌 시간대에 맞춰 윈도우를 조정하는 거예요. 예를 들어 중국에서는 Duration.ofHours(-8) 오프셋을 지정해야 해요.

슬라이딩 윈도우 (Sliding Windows)

슬라이딩 윈도우 어사이너는 요소를 고정 길이의 윈도우에 배정해요. 텀블링과 비슷하게 윈도우 크기(window size) 파라미터로 크기를 설정하고, 추가로 슬라이드(window slide) 파라미터가 슬라이딩 윈도우가 얼마나 자주 시작되는지를 제어해요. 그래서 슬라이드가 윈도우 크기보다 작으면 슬라이딩 윈도우는 겹칠 수 있어요. 이 경우 요소는 여러 윈도우에 배정되죠.

예를 들어 크기 10분, 슬라이드 5분의 윈도우를 만들면, 5분마다 지난 10분 동안 도착한 이벤트를 담은 윈도우가 생겨요.

DataStream<T> input = ...;

input
    .keyBy(<key selector>)
    .window(SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofSeconds(5)))
    .<windowed transformation>(<window function>);

input \
    .key_by(<key selector>) \
    .window(SlidingEventTimeWindows.of(Duration.ofSeconds(10), Duration.ofSeconds(5))) \
    .<windowed transformation>(<window function>)

세션 윈도우 (Session Windows)

세션 윈도우는 **활동이 없는 간격(gap)**으로 묶어요. 앞의 두 윈도우와 달리 세션 윈도우는 고정 길이를 갖지 않고, 활동이 없어 잘리는 시간에 따라 경계가 정해져요. 어떤 사용자가 활발하게 이용하다가 잠시 활동을 멈추고 다시 이용하면, 세션 윈도우 어사이너는 그 사이의 갭을 기준으로 두 활동을 서로 다른 세션 윈도우에 넣어요. 여기서 '활동 없음'이란 처리 시간 또는 이벤트 시간으로 정의됩니다.

늦은 데이터 다루기 (Allowed Lateness)

기본적으로 워터마크가 윈도우의 종료 타임스탬프를 지나가면 해당 요소는 버려져요. 하지만 윈도우 어사이너에 걸리는 시간(event time) 조건과 함께 allowedLateness(...)를 지정하면, 늦게 도착해 버려지던 데이터를 윈도우가 계속 받아주도록 할 수 있어요. allowed lateness가 0보다 크면 워터마크가 윈도우 끝을 지나간 뒤에도 윈도우와 내용물이 유지돼요. 이때 늦게 도착했지만 버려지지 않은 요소가 오면 윈도우가 다시 발화(trigger)할 수 있는데, 이것을 late firing이라 불러요. 최초의 발화(main firing)와 대비되는 개념이죠. 세션 윈도우의 경우 late firing이 두 윈도우 사이의 갭을 '이어붙여' 머지되지 않은 기존 윈도우들을 합칠 수도 있어요.

late firing으로 내보낸 요소는 이전 계산의 갱신된 결과로 취급해야 해요. 즉 데이터 스트림에 같은 계산의 결과가 여러 개 들어올 수 있다는 뜻이에요. 애플리케이션에 따라 이 중복 결과를 감안하거나 중복 제거해야 해요.

늦은 데이터를 사이드 출력으로 받기

Flink의 사이드 출력(side output) 기능을 쓰면 늦어서 버려지는 데이터를 스트림으로 받아올 수 있어요. 먼저 윈도우 스트림에 sideOutputLateData(OutputTag)를 지정하고, 윈도우 연산 결과에서 getSideOutput(OutputTag)로 사이드 출력 스트림을 얻는 식이에요.

final OutputTag<T> lateOutputTag = new OutputTag<T>("late-data"){};

DataStream<T> input = ...;

SingleOutputStreamOperator<T> result =
    input
        .keyBy(<key selector>)
        .window(<window assigner>)
        .allowedLateness(<time>)
        .sideOutputLateData(lateOutputTag)
        .<windowed transformation>(<window function>);

DataStream<T> lateStream = result.getSideOutput(lateOutputTag);

late_output_tag = OutputTag("late-data", type_info)

result = input \
    .key_by(<key selector>) \
    .window(<window assigner>) \
    .allowed_lateness(<time>) \
    .side_output_late_data(late_output_tag) \
    .<windowed transformation>(<window function>)

late_stream = result.get_side_output(late_output_tag)

윈도우 결과 다루기

윈도우 연산의 결과는 다시 DataStream이에요. 결과 요소에는 윈도우 연산에 대한 정보가 남지 않으므로, 윈도우에 대한 메타정보를 유지하고 싶다면 ProcessWindowFunction에서 결과 요소에 직접 인코딩해야 해요. 결과 요소에 설정되는 유일한 관련 정보는 요소 타임스탬프인데, 처리된 윈도우의 허용 최대 타임스탬프(종료 타임스탬프 - 1, 종료는 제외라서)로 설정돼요. 이것은 이벤트 시간 윈도우와 처리 시간 윈도우 모두에 해당돼요.

워터마크와 윈도우의 상호작용

이벤트 시간 윈도우 연산을 할 때 워터마크는 윈도우가 닫히고 발화할 시점을 결정하는 데 결정적 역할을 해요. 요약하면, 워터마크가 윈도우의 종료 타임스탬프(앞서 본 allowed lateness를 더한 시점 포함)까지 지나간 뒤, 그 윈도우에 도착하는 이벤트는 다음 두 가지 중 하나로 처리돼요: (a) 윈도우가 아직 열려 있으면(수정/추가 가능한 경우) 윈도우에 추가되거나, (b) 워터마크가 이미 지나가 버려졌으면 늦은 데이터로 취급돼요. late firing과 allowed lateness는 바로 이 지점에서 의미를 가져요.

더 알아보기 (Learn more)