6장: Apache Flink — 스트림 처리 엔진

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
 
public class OrderAnalytics {
    public static void main(String[] args) throws Exception {
        // 1. 실행 환경 생성
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
 
        // 2. 소스에서 데이터 읽기
        DataStream<String> rawStream = env.addSource(
            new FlinkKafkaConsumer<>("orders", new SimpleStringSchema(), kafkaProps)
        );
 
        // 3. 변환 처리
        DataStream<OrderEvent> orders = rawStream
            .map(json -> OrderEvent.fromJson(json))
            .filter(order -> order.getAmount() > 0);
 
        // 4. 키 기반 집계
        DataStream<OrderSummary> summary = orders
            .keyBy(order -> order.getUserId())
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(new OrderAggregator());
 
        // 5. 싱크에 결과 쓰기
        summary.addSink(new ElasticsearchSink<>(esConfig));
 
        // 6. 실행
        env.execute("Order Analytics Job");
    }
}

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common import Time
 
# 1. 실행 환경 생성
env = StreamExecutionEnvironment.get_execution_environment()
 
# 2. 소스에서 데이터 읽기
raw_stream = env.add_source(kafka_source)
 
# 3. 변환 처리
orders = (raw_stream
    .map(parse_order)
    .filter(lambda order: order.amount > 0))
 
# 4. 키 기반 집계
summary = (orders
    .key_by(lambda order: order.user_id)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(OrderAggregator()))
 
# 5. 싱크에 결과 쓰기
summary.add_sink(es_sink)
 
# 6. 실행
env.execute("Order Analytics Job")

// 텀블링 윈도우: 5분 단위 매출 집계
orders
    .keyBy(order -> order.getCategory())
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .sum("amount");
 
// 슬라이딩 윈도우: 10분 윈도우를 1분 간격으로 슬라이드
orders
    .keyBy(order -> order.getCategory())
    .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
    .sum("amount");
 
// 세션 윈도우: 30분 비활동 시 세션 종료
userActions
    .keyBy(action -> action.getUserId())
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .process(new SessionAnalyzer());

// 윈도우 조인: 주문과 결제를 5분 윈도우 내에서 조인
DataStream<EnrichedOrder> enriched = orders
    .join(payments)
    .where(order -> order.getOrderId())
    .equalTo(payment -> payment.getOrderId())
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .apply((order, payment) -> new EnrichedOrder(order, payment));

// 주문 후 1시간 이내의 결제만 조인
orders
    .keyBy(order -> order.getOrderId())
    .intervalJoin(payments.keyBy(payment -> payment.getOrderId()))
    .between(Time.minutes(0), Time.hours(1))
    .process(new OrderPaymentJoiner());

public class FraudDetector extends KeyedProcessFunction<String, Transaction, Alert> {
 
    // ValueState: 단일 값 저장
    private ValueState<Boolean> flagState;
 
    // ListState: 값 목록 저장
    private ListState<Transaction> recentTransactions;
 
    // MapState: 키-값 매핑 저장
    private MapState<String, Integer> merchantCounts;
 
    @Override
    public void open(Configuration parameters) {
        flagState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("flag", Boolean.class));
        recentTransactions = getRuntimeContext().getListState(
            new ListStateDescriptor<>("recent", Transaction.class));
        merchantCounts = getRuntimeContext().getMapState(
            new MapStateDescriptor<>("merchants", String.class, Integer.class));
    }
 
    @Override
    public void processElement(Transaction tx, Context ctx, Collector<Alert> out) {
        // 상태를 활용한 사기 탐지 로직
        Boolean flagged = flagState.value();
        if (flagged != null && flagged) {
            if (tx.getAmount() > 500) {
                out.collect(new Alert(tx.getUserId(), "고액 거래 감지"));
            }
        }
        if (tx.getAmount() < 1) {
            flagState.update(true); // 소액 테스트 거래 감지
            // 다음 이벤트에서 검증
            ctx.timerService().registerEventTimeTimer(
                ctx.timestamp() + 60_000); // 1분 후 플래그 초기화
        }
    }
 
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
        flagState.clear(); // 타이머 만료 시 플래그 초기화
    }
}

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
 
// 대규모 상태가 필요한 경우 RocksDB 사용
env.setStateBackend(new EmbeddedRocksDBStateBackend());

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();
 
// 체크포인트 활성화: 30초 간격
env.enableCheckpointing(30_000);
 
// Exactly-once 보장 (기본값)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
 
// 체크포인트 간 최소 간격: 10초
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10_000);
 
// 체크포인트 타임아웃: 60초
env.getCheckpointConfig().setCheckpointTimeout(60_000);
 
// 체크포인트 저장소 (HDFS, S3 등)
env.getCheckpointConfig().setCheckpointStorage("s3://flink-checkpoints/");

-- 1. Kafka 소스 테이블 정의
CREATE TABLE orders (
    order_id STRING,
    user_id STRING,
    amount DECIMAL(10, 2),
    category STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'flink-sql-orders',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset'
);
 
-- 2. 5분 단위 카테고리별 매출 집계
CREATE TABLE category_revenue (
    category STRING,
    window_start TIMESTAMP(3),
    window_end TIMESTAMP(3),
    total_revenue DECIMAL(10, 2),
    order_count BIGINT,
    PRIMARY KEY (category, window_start) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-7',
    'hosts' = 'http://elasticsearch:9200',
    'index' = 'category-revenue'
);
 
INSERT INTO category_revenue
SELECT
    category,
    window_start,
    window_end,
    SUM(amount) AS total_revenue,
    COUNT(*) AS order_count
FROM TABLE(
    TUMBLE(TABLE orders, DESCRIPTOR(order_time), INTERVAL '5' MINUTE)
)
GROUP BY category, window_start, window_end;