Triggers

상위: Flink

요약

Flink의 Trigger 메커니즘을 다룹니다. 윈도우가 언제 결과를 출력할지 결정하는 조건과 규칙을 설정하는 방법을 학습하고, 커스텀 트리거를 구현하여 특수한 요구사항을 만족하는 윈도우 처리를 구현하는 방법을 알아봅니다.

Trigger

• 윈도우가 언제 결과를 내보낼지 결정하는 조건/규칙.

• 각 윈도우에는 디폴트 트리거가 정의되어 있어, 특별히 지정하지 않아도 동작함.

• 기본 트리거 종류

  • 이벤트 시간 기반 윈도우
  • 처리 시간 기반 윈도우
  • 세션 윈도우

시간 기반 트리거

• 일정 시간 간격으로 윈도우를 강제로 출력.
• 예: Processing-Time Trigger: 5초마다 강제 출력

카운트 기반 트리거

• 윈도우에 누적된 이벤트 개수가 특정 숫자에 도달할 때 출력
• 예: 100개 이벤트가 모일 때마다 출력

복합 조건 트리거

• 시간 또는 카운트 등 여러 조건 중 하나 만족 시 출력하거나, AND 조건도 조합 가능
• 예: 1분이 지났거나 50개가 모이면 출력

사용자 정의 임의 조건

• 특정 값 이상이 되면 출력
• Trigger 인터페이스 구현하여 onElement()에서 조건 만족 시 FIRE

TriggerResult 종류

기본 Trigger 유형

• ProcessingTimeTrigger: 처리 시간 기준 특정 시점에 FIRE
• EventTimeTrigger: 이벤트 시간 워터마크 기준 FIRE (모든 이벤트-타임 윈도우의 디폴트)
• CountTrigger: 누적 이벤트 개수가 기준 도달 시 FIRE

CustomTrigger

Trigger 클래스 상속 & 필수 메서드 구현

• 반드시 5개 메서드 구현 필요 (미사용 메서드는 TriggerResult.CONTINUE로 처리)

CustomTrigger 예제

• Trigger 클래스 상속 및 구현 예시

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

    @staticmethod
    def of(count_threshold):
        return CustomCountTrigger(count_threshold)

    def on_element(self, element, timestamp, window, ctx):
        count_state_desc = ValueStateDescriptor("count", Types.INT())
        count_state = ctx.get_partitioned_state(count_state_desc)

        current_count = count_state.value() or 0
        current_count += 1
        count_state.update(current_count)

        if current_count >= self.count_threshold:
            count_state.clear()
            return TriggerResult.FIRE_AND_PURGE
        return TriggerResult.CONTINUE

# 실행 환경 설정
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

# 입력 데이터 생성
data = [(i,) for i in range(1, 16)]
ds = env.from_collection(collection=data, type_info=Types.TUPLE([Types.INT()]))

# CountTrigger(5) 적용: 5개마다 실행
windowed = ds.window_all(GlobalWindows.create()) \
    .trigger(CustomCountTrigger.of(5)) \
    .reduce(lambda a, b: (a[0] + b[0],))

# 결과 출력
windowed.print()
env.execute("Custom CountTrigger Example")

출력 :

(15)
(40)
(65)

고급 트리거 패턴

1. 시간 + 카운트 복합 트리거

class TimeOrCountTrigger(Trigger):
    def __init__(self, time_threshold, count_threshold):
        self.time_threshold = time_threshold
        self.count_threshold = count_threshold

    def on_element(self, element, timestamp, window, ctx):
        # 카운트 체크
        count_state = ctx.get_partitioned_state(
            ValueStateDescriptor("count", Types.INT())
        )
        current_count = count_state.value() or 0
        current_count += 1
        count_state.update(current_count)

        if current_count >= self.count_threshold:
            return TriggerResult.FIRE_AND_PURGE
        
        # 타이머 설정
        ctx.register_processing_time_timer(
            ctx.get_current_processing_time() + self.time_threshold
        )
        
        return TriggerResult.CONTINUE

    def on_processing_time(self, timestamp, window, ctx):
        return TriggerResult.FIRE_AND_PURGE

2. 조건부 트리거

class ConditionalTrigger(Trigger):
    def __init__(self, condition_func):
        self.condition_func = condition_func

    def on_element(self, element, timestamp, window, ctx):
        if self.condition_func(element):
            return TriggerResult.FIRE_AND_PURGE
        return TriggerResult.CONTINUE

3. 상태 기반 트리거

class StateBasedTrigger(Trigger):
    def __init__(self, state_key, threshold):
        self.state_key = state_key
        self.threshold = threshold

    def on_element(self, element, timestamp, window, ctx):
        state = ctx.get_partitioned_state(
            ValueStateDescriptor(self.state_key, Types.DOUBLE())
        )
        
        current_value = state.value() or 0.0
        new_value = current_value + element[1]  # 예: 누적 합계
        
        state.update(new_value)
        
        if new_value >= self.threshold:
            state.clear()
            return TriggerResult.FIRE_AND_PURGE
        
        return TriggerResult.CONTINUE

트리거 최적화

1. 상태 관리 최적화

class OptimizedTrigger(Trigger):
    def __init__(self):
        self.state_descriptor = ValueStateDescriptor("trigger_state", Types.INT())

    def on_element(self, element, timestamp, window, ctx):
        # 상태 재사용으로 성능 향상
        state = ctx.get_partitioned_state(self.state_descriptor)
        # ... 로직 처리

2. 타이머 최적화

def on_element(self, element, timestamp, window, ctx):
    # 불필요한 타이머 등록 방지
    if not self.timer_registered:
        ctx.register_processing_time_timer(timestamp + self.interval)
        self.timer_registered = True

3. 메모리 정리

def clear(self, window, ctx):
    # 상태 정리
    state = ctx.get_partitioned_state(self.state_descriptor)
    state.clear()
    
    # 타이머 정리
    ctx.delete_processing_time_timer(self.timer_timestamp)

관련 개념

• Window Implementation - 윈도우 구현 방법
• Evictors - 윈도우 데이터 제거
• Watermarks - 이벤트 시간 처리