Apache Flink

# PyFlink 스트림 처리 예제

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import (
    KafkaSource, KafkaOffsetsInitializer, KafkaSink, KafkaRecordSerializationSchema
)
from pyflink.datastream.functions import MapFunction, KeyedProcessFunction, RuntimeContext
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.common import Types, WatermarkStrategy, Duration
from pyflink.common.serialization import SimpleStringSchema
import json

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

# 체크포인팅 설정 (exactly-once 보장)
env.enable_checkpointing(60000)  # 60초마다 체크포인트
env.get_checkpoint_config().set_min_pause_between_checkpoints(30000)

# ==========================================
# 1. Kafka Source 설정
# ==========================================

kafka_source = KafkaSource.builder() \
    .set_bootstrap_servers("kafka:9092") \
    .set_topics("user-events") \
    .set_group_id("flink-processor") \
    .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

# 워터마크 전략 (이벤트 시간 기반)
watermark_strategy = WatermarkStrategy \
    .for_bounded_out_of_orderness(Duration.of_seconds(10)) \
    .with_timestamp_assigner(lambda event, ts: json.loads(event)['timestamp'])

# 소스 스트림 생성
events = env.from_source(
    kafka_source,
    watermark_strategy,
    "Kafka Source"
)

# ==========================================
# 2. 사용자 세션 분석 (Stateful Processing)
# ==========================================

class SessionAggregator(KeyedProcessFunction):
    """사용자별 세션 집계 (상태 유지)"""

    def open(self, runtime_context: RuntimeContext):
        # 상태 초기화
        self.session_state = runtime_context.get_state(
            ValueStateDescriptor("session", Types.STRING())
        )
        self.event_count = runtime_context.get_state(
            ValueStateDescriptor("count", Types.INT())
        )

    def process_element(self, event, ctx):
        data = json.loads(event)
        user_id = data['user_id']

        # 현재 상태 가져오기
        current_session = self.session_state.value()
        count = self.event_count.value() or 0

        # 세션 타임아웃 타이머 등록 (30분)
        ctx.timer_service().register_event_time_timer(
            ctx.timestamp() + 30 * 60 * 1000
        )

        # 상태 업데이트
        self.event_count.update(count + 1)

        if current_session is None:
            # 새 세션 시작
            self.session_state.update(data.get('session_id'))
            yield json.dumps({
                'user_id': user_id,
                'event': 'session_start',
                'timestamp': data['timestamp']
            })

        yield json.dumps({
            'user_id': user_id,
            'event_type': data['event_type'],
            'session_events': count + 1
        })

    def on_timer(self, timestamp, ctx):
        # 세션 종료
        count = self.event_count.value()
        yield json.dumps({
            'user_id': ctx.get_current_key(),
            'event': 'session_end',
            'total_events': count,
            'timestamp': timestamp
        })
        # 상태 클리어
        self.session_state.clear()
        self.event_count.clear()

# 스트림 처리 파이프라인
processed = events \
    .map(lambda x: json.loads(x)) \
    .key_by(lambda x: x['user_id']) \
    .process(SessionAggregator())

# ==========================================
# 3. Flink SQL로 실시간 집계
# ==========================================

from pyflink.table import StreamTableEnvironment, EnvironmentSettings

# Table 환경 설정
settings = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, settings)

# Kafka 테이블 정의
t_env.execute_sql("""
    CREATE TABLE user_events (
        user_id STRING,
        event_type STRING,
        amount DECIMAL(10, 2),
        event_time TIMESTAMP(3),
        WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'user-events',
        'properties.bootstrap.servers' = 'kafka:9092',
        'properties.group.id' = 'flink-sql',
        'format' = 'json',
        'scan.startup.mode' = 'latest-offset'
    )
""")

# 윈도우 집계 쿼리
t_env.execute_sql("""
    SELECT
        user_id,
        TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,
        COUNT(*) AS event_count,
        SUM(amount) AS total_amount
    FROM user_events
    WHERE event_type = 'purchase'
    GROUP BY
        user_id,
        TUMBLE(event_time, INTERVAL '1' MINUTE)
""")

# 실행
env.execute("User Session Analysis")