Window Implementation

상위: Flink

요약

Flink에서 다양한 윈도우 타입을 구현하는 방법을 다룹니다. 텀블링, 슬라이딩, 세션, 글로벌 윈도우의 실제 코드 구현과 실행 결과를 통해 윈도우 처리의 동작 원리를 학습합니다.

공통 환경 설정

import time
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.common.time import Time
from pyflink.datastream.window import SlidingProcessingTimeWindows
from pyflink.common.typeinfo import Types

# 실행 환경 생성
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

# 데이터 소스 생성
data_stream = env.from_collection(
    collection=[("user1", 1), ("user1", 3), ("user1", 4), ("user1", 5)],
    type_info=Types.TUPLE([Types.STRING(), Types.INT()])
)

슬라이딩 윈도우 (Sliding Window)

# 1.2초 지연 추가
delayed_stream = data_stream.map(lambda x: (x[0], x[1]) if time.sleep(1.2) is None else (x[0], x[1]),
                                 output_type=Types.TUPLE([Types.STRING(), Types.INT()]))

# 슬라이딩 윈도우 적용: 크기 2초, 슬라이드 간격 1초
windowed_stream = delayed_stream \
    .key_by(lambda x: x[0]) \
    .window(SlidingProcessingTimeWindows.of(Time.seconds(2), Time.seconds(1))) \
    .reduce(lambda a, b: (a[0], a[1] + b[1]))

windowed_stream.print()
env.execute("ProcessingTime Window Example")

출력 :

(user1, 1)
(user1, 4)
(user1, 7)
(user1, 9)

세션 윈도우 (Session Window)

from pyflink.datastream.window import ProcessingTimeSessionWindows

# 지연을 위한 함수 정의
def delayed_map(record):
    if record[1] == 0:
        time.sleep(0.5)
    elif record[1] in [1, 3]:
        time.sleep(2.0)
    elif record[1] == 4:
        time.sleep(2.8)
    elif record[1] == 5:
        time.sleep(0.5)
    return record

delayed_stream = data_stream.map(delayed_map, output_type=Types.TUPLE([Types.STRING(), Types.INT()]))

# 세션 윈도우 적용: 세션 간격 2초
windowed_stream = delayed_stream \
    .key_by(lambda x: x[0]) \
    .window(ProcessingTimeSessionWindows.with_gap(Time.seconds(2))) \
    .reduce(lambda a, b: (a[0], a[1] + b[1]))

windowed_stream.print()
env.execute("Session Window Split Example")

출력 :

(user1, 4)
(user1, 9)

글로벌 윈도우 (Global Window)

from pyflink.datastream.window import GlobalWindows, Trigger, TriggerResult
from pyflink.datastream.state import ValueStateDescriptor

class CustomCountTrigger(Trigger):
    def __init__(self, count):
        self.count = count

    def on_element(self, element, timestamp, window, ctx):
        state = ctx.get_partitioned_state(ValueStateDescriptor("count", Types.INT()))
        current = state.value() or 0
        current += 1
        if current >= self.count:
            state.clear()
            return TriggerResult.FIRE_AND_PURGE
        else:
            state.update(current)
            return TriggerResult.CONTINUE

data = [1, 2, 3, 4, 5, 6]
ds = env.from_collection(collection=data, type_info=Types.INT())

# 글로벌 윈도우 적용 및 커스텀 트리거 설정
windowed = ds.window_all(GlobalWindows.create()) \
    .trigger(CustomCountTrigger.of(3)) \
    .reduce(lambda a, b: a + b)

windowed.print()
env.execute("GlobalWindows Example")

출력 :

6
15

텀블링 윈도우 (Tumbling Window)

from pyflink.datastream.window import TumblingProcessingTimeWindows

delayed_stream = data_stream.map(delayed_map, output_type=Types.TUPLE([Types.STRING(), Types.INT()]))

# 텀블링 윈도우 적용: 2초 간격
windowed_stream = delayed_stream \
    .key_by(lambda x: x[0]) \
    .window(TumblingProcessingTimeWindows.of(Time.seconds(2))) \
    .reduce(lambda a, b: (a[0], a[1] + b[1]))

windowed_stream.print()
env.execute("Tumbling Window Example")

출력 :

(user1, 4)
(user1, 9)

윈도우 구현 패턴

1. 기본 윈도우 설정

# 텀블링 윈도우
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))

# 슬라이딩 윈도우
.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))

# 세션 윈도우
.window(ProcessingTimeSessionWindows.with_gap(Time.seconds(30)))

2. 윈도우 함수 적용

# Reduce 함수
.reduce(lambda a, b: (a[0], a[1] + b[1]))

# Aggregate 함수
.aggregate(MyAggregateFunction())

# Process 함수
.process(MyProcessWindowFunction())

3. 키 기반 윈도우

# 키별로 윈도우 적용
stream.key_by(lambda x: x[0]) \
      .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) \
      .reduce(lambda a, b: (a[0], a[1] + b[1]))

윈도우 최적화 기법

1. 적절한 윈도우 크기 선택

# 너무 작은 윈도우: 오버헤드 증가
.window(TumblingProcessingTimeWindows.of(Time.milliseconds(100)))

# 너무 큰 윈도우: 메모리 사용량 증가
.window(TumblingProcessingTimeWindows.of(Time.hours(24)))

# 적절한 윈도우 크기
.window(TumblingProcessingTimeWindows.of(Time.minutes(5)))

2. 윈도우 함수 최적화

# 효율적인 Reduce 함수
def efficient_reduce(a, b):
    return (a[0], a[1] + b[1])  # 단순한 덧셈

# 비효율적인 Process 함수 (복잡한 로직)
def complex_process(window, elements):
    # 복잡한 계산...
    pass

3. 메모리 관리

# 윈도우 상태 정리
class MyProcessWindowFunction(ProcessWindowFunction):
    def process(self, key, window, context, elements):
        # 윈도우 처리
        result = self.process_elements(elements)
        
        # 상태 정리
        context.window_state().clear()
        
        return result

관련 개념

• Window Concepts - 윈도우 개념 및 종류
• Triggers - 윈도우 트리거 메커니즘
• Watermarks - 이벤트 시간 처리