로컬에서 스트리밍 파이프라인 올려보기 with AI

지금까지의 데이터 파이프라인 경험은 배치(Batch) 기반의 ETL이었다. Airflow, Hive, Spark 등을 활용해 정해진 주기로 데이터를 처리하는 방식에 익숙했다.

상대적으로 실시간 데이터 처리는 깊게 다뤄볼 기회가 없었다. 하지만 데이터 엔지니어의 역량은 배치 처리뿐 아니라 스트리밍 처리도 필수적이고, 요즘은 실시간 처리가 중요하다고 느껴서 관련 스택을 제대로 공부해야겠다는 생각이 들었다.

Kafka, 스트리밍 엔진, 스토리지 및 OLAP 엔진 등에 대해서도 공부한 내용들도 포스팅해야하지만, 로컬에서 간단하게 올려봐야겠다 생각되어 Claude & Gemini CLI 두 시니어 엔지니어분들와 함께 간단한 유즈케이스로 로컬에서 (다행히) 돌아가는 스트리밍 파이프라인을 만들어 금번 포스팅에는 해당 내용을 기재하고자 한다.

요구사항

나만의 요구사항은 다음과 같았다.

광고 노출과 클릭 이벤트를 실시간으로 받아서, 10초마다 CTR을 집계하고 이를 API로 서빙한다.
처리 시간이 아닌 이벤트 시간을 기준으로 삼아야한다.
이벤트는 At least once를 준수한다.
CTR의 변경사항을 2초이내에 반영할 수 있어야한다.
API를 통해 집계 결과를 받을 수 있어야하고, 조회 시간은 1s을 넘지 않아야한다.

부가적인 요구사항으론 다음을 가져갔다.

각각의 흐름은 직접 확인할 수 있게 Web UI를 제공한다

하드웨어 환경은 다음과 같았다.

M3 MAX
36GB Memory

테스트 결과

결과부터 먼저 확인하면, 맥북 프로 구매 후 처음으로 나만의 고양이를 혹사 시켜볼 수 있었다.

우선 프로듀서를 감으로 셋업한 다음, Flink UI를 확인했을 때, 초당 4K정도의 이벤트가 발생하는 것을 확인할 수 있었다. 약 25분정도 넉넉하게 돌려보았고, 6M 정도의 이벤트가 Flink에서 처리되었다.

아무리 로컬이지만, Flink 병렬도도 2로밖에 안잡은 상황인데, 두 토픽의 총 6개의 파티션 오프셋이 그때그때 바로 처리되는 것을 확인할 수 있었다.

서빙 API 성능은 1s를 만족했고, 100ms 단위로 집계 결과가 변경되는 것을 확인할 수 있었다.
로컬이라서 Network I/O가 매우 적은 환경이었다.

이러한 수치들을 확인하며, 처음 목표했었던 10초 단위 집계를 2초 안에 받아야 하고, API 성능은 1초를 넘기지 않아야한다는 나만의 성능 측면의 요구사항은 달성했다고 생각했다.

다만, 별도로 스트림 이벤트의 설정은 진행되지 않아 at least once가 완전하게 보장하진 못했다.

Design

구성한 파이프라인은 다음과 같다.

Flink는 JobManager가 웹 UI를 제공하고, Kafka랑 Redis는 눈으로 직접 쌓이는걸 보기 위해 각각 provectuslabs/kafka-ui와 redis/redisinsight를 사용했다.

전체 데이터 흐름

이벤트 생성부터 API 응답까지의 전체 흐름은,
Impression과 Click 이벤트가 각각 0.001초, 0.002초 간격으로 생성되어 Kafka 토픽으로 전송된다.
이후 Flink가 두 토픽의 데이터를 소비해서 10초 윈도우 단위로 CTR을 계산하고, 결과를 Redis에 저장한다.
마지막으로 FastAPI가 Redis에서 데이터를 읽어 REST API로 제공하는 흐름이다.

프로듀서를 제외한 모든 컴포넌트들은 docker-compose를 통해 셋업되었다.

kafka

Kafka는 3개의 브로커로 구성된 클러스터를 사용했다. Zookeeper는 1개로 구성했다.
토픽은 impressions, clicks 2개를 사용하고 각각 3개의 파티션으로 구성했다.
토픽별로 3개의 파티션을 할당한 이유는, 병렬 처리 성능을 확보하기 위함이었다.
추가적으로 Producer App에서product_id를 기준으로 파티셔닝하여, 동일한 상품의 이벤트들이 같은 파티션에 퍼블리싱되게 구성하였다.

Flink

이번 구성에서 실시간 처리의 핵심 컴포넌트이다. 로컬 환경에서는 병렬도를 2로 설정해서 사용했다.

핵심 설정:

윈도우: 10초 Tumbling Window
시간 기준: Event Time
워터마크: 2초 (다이어그램에는 2초로 표시되어 있지만, 실제 테스트에서는 조정 가능)
Allowed Lateness: 5초

Event Time 기반 처리를 요구사항에 포함한 이유는 CTR 집계 같은 상황은 실제 이벤트 발생 시간 기준으로 이루어진다 생각했기 때문이다. Processing Time을 사용하면 네트워크 지연이나 시스템 부하로 인해 늦게 도착하는 이벤트들이 제외되는 상황이 발생할 수 있기 때문에, 실시간이지만 정확성을 높이기 위해선 Event Time 기준이 적절하다 판단하였다.

Event Time을 사용하였으므로, 집계 시작 시간을 지정하기 위해 WaterMark를 2초로 설정하였고, Allowed Lateness는 5초를 할당하였다.

Flink 말고도 Spark Streaming, Kafka Streams를 사용할 수 있었지만, 각각 다음 사유로 해당 유즈케이스에선 Flink가 적합하다 판단했다.

공통 가정 : 로컬 환경이어 부하가 크진 않지만, 부하가 있는 상황을 가정하여 가능한 최적의 유즈케이스를 고려한다.

Kafka Streams : 집계가 포함되어 있기 때문에, Kafka 부하를 줄이고자 선택에서 제외하였다. Kafka Streams 집계나 조인 연산 시, 파티션 셔플링이 필요한 상황을 내부 토픽 생성을 통해 이벤트들을 별도로 모아 처리하는 방식인데, 이 과정이 Kafka에서 일어나기 때문에 Kafka를 이벤트 플랫폼의 역할에 집중하고, 스트림 처리는 별도로 구성하기 위해 제외하였다.

Spark Streams : 집계가 포함되어 있기 때문에 Continous 모드는 사용할 수 없고 마이크로 배치 모드로 사용해야한다. 지연 시간을 고려했을 때 1s ~ 2s 내에 업데이트 요구사항은 만족할 수 있다 판단되지만, 실제 환경 + 요구사항이 100ms 이내로 변경되는 상황을 고려한다면 Flink가 더 적절하다 판단했다.

Redis

집계 결과는 Redis에 저장했다. In-memory 저장소이기 때문에 매우 빠른 읽기 성능을 제공하여, 집계 결과를 빠르게 서빙하기 위한 저장소로써 활용했다.

저장 구조

ctr:latest - 최신 윈도우의 CTR 결과 (Hash)
ctr:previous - 이전 윈도우의 CTR 결과 (Hash)

각 Hash의 field는 상품 ID이고, value는 JSON 형태의 CTR 데이터다. 이전 윈도우 결과까지 보관하는 케이스를 상정했다.

{
  "product_123": "{\"impressions\": 1500, \"clicks\": 45, \"ctr\": 0.03, \"timestamp\": \"2024-01-15T10:30:00Z\"}"
}

FastAPI

FastAPI는 Redis에서 데이터를 읽어 RESTful API로 제공한다. 간단한 API 몇개만 구성할 생각이였어서 이를 선택했다.

엔드포인트:

/ctr/{product_id} : 특정 상품의 현재/이전 CTR 비교
/ctr/latest : 모든 상품의 최신 CTR
/ctr/previous : 모든 상품의 이전 CTR

@app.get("/ctr/latest", summary="Get Latest CTR for All Products")  
def get_all_latest_ctr():  
	ctr_hash = redis_client.hgetall("ctr:latest")  
	if not ctr_hash:  
		return {}  
	result = {key: json.loads(value) for key, value in ctr_hash.items()}  
	return result

Flink 윈도우 처리의 핵심

스트리밍 처리에서 가장 중요한 부분은 시간 윈도우 설정이었다. CTR 계산을 위해 Event Time을 사용하면서 WaterMark와 Lateness를 설정해야했다.

ProcessTime : 엔진에 들어온 시간 순서대로 처리한 후 집계하는 방식. 지정한 윈도우 범위마다 바로바로 연산이 수행된다.
EventTime : 이벤트의 타임스탬프 기준을 고려하여 집계하는 방식. 윈도우범위와 함께 연산을 시작할 워터마크를 지정해야한다.

추가적으로 10초마다 CTR을 집계하는 요구사항을 만족하기 Tumbling Window 타입을 사용했다.
이는 겹치지 않는 고정 크기의 윈도우이다. 10분 윈도우라면 00:00-00:10, 00:10-00:20 처럼 딱딱 구분되어 진행되는 방식이다.

Sliding Window는 겹치는 윈도우를 만들 수 있는 타입으로, 10분 윈도우를 5분마다 생성한다면 00:00-00:10, 00:05-00:15, 00:10-00:20 이런 식으로 생성할 수 있는 방식이다.

Session Window는 이벤트 간의 비활성 시간을 기준으로 윈도우를 생성한다. 예를 들어, 30분의 세션 타임아웃을 설정하면, 사용자가 30분 동안 활동이 없으면 세션이 종료되고 새로운 세션이 시작된다.

WatermarkStrategy<Event> watermarkStrategy = WatermarkStrategy  
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(2))  
        .withTimestampAssigner(new EventTimestampExtractor());  
  
SingleOutputStreamOperator<CTRResult> ctrResults = combinedStream  
        .keyBy(new ProductKeySelector())  
        .window(TumblingEventTimeWindows.of(Time.seconds(10)))  
        .allowedLateness(Time.seconds(5))  
        .aggregate(new EventCountAggregator(), new CTRResultWindowProcessFunction());

현재 한계점 - 만약 운영 환경이라면?

운영 환경을 염두했을 때, 고려해야할 점은 크게 3가지일 것 같다.

1. ThroughtPut
만약 이벤트가 초당 4K가 아닌 10K 이상으로 증가하고, 이를 대응하기위해선 토픽 파티션을 증가시키고, Flink 병렬도와 세부 설정을 통해 대응할 수 있지 않을까 생각된다.
또한, 상품 개수가 수십만건일 경우에는 단일 노드가 아닌 Cluster로 레디스를 구성하고, 분산 키 전략을 사용해 노드의 메모리 사용량을 분산할 수 있다.

2. Failover 대응
현재는 장애가 발생할 포인트가 명확한데, Kafka 를 제외하면 Flink & Redis & FastAPI 모두가 SPOF이다. 이는 K8s 클러스터의 오케스트레이션을 사용하면 파드에 대한 복구는 가능할 것으로 예상되지만, 스트리밍의 처리 흐름을 복구하기 위해선 Kafka, Flink 옵션을 세부적으로 조절해야할 것 같다.
Kafka 브로커를 늘리고, 파티션 복제 수준인 replication-factor와 쓰기 허용 기준인 min.insync.replicas을 조절할 수 있을 것 같다.
클러스터 자체의 장애가 발생하는 것을 대비하기 위해선 여분의 클러스터를 두어 미러링하는 방법이 존재한다.

Flink Checkpoint 설정을 진행하지 않았다. 하지만 장애 상황에서 스트림을 복구하고 Exactly-Once 전달을 보장하기 위해서는 Checkpoint가 핵심 기능이므로, 운영 환경에서는 이 부분에 대한 추가 설정이 필요할 것 같다. (추가적으로 학습이 필요한 부분)

Redis 데이터가 휘발되는 상황을 고려했을 때는 Replication을 구성해 Slave Node를 별도로 두고, Redis가 아닌 HDFS, S3같은 별도 저장소에 집계 데이터를 따로 보관이 필요할 것 같다.
파일을 직접 읽어야하니 성능은 매우 느리겠지만, API의 Key Miss가 발생했을 때 백업 플랜으로 Hive나 Iceberg에서 데이터를 조회하는 방안이 마련되어야한다.

3. 모니터링
금번 작업에선 Web UI를 붙여두어서 Kafka, Flink, Redis, FastAPI에 대한 기본적인 모니터링은 가능하지만, 각 컨테이너의 메트릭 & 로깅을 위한 모니터링은 구성되지 않았다. 자주 사용되는 조합인 Prometheus & Grafana // Fluentd & ES & Kibana 조합으로 구성할 수 있을 것 같다.

마무리

금번 작업 내용은 다음 Repo에서 확인할 수 있다.
https://github.com/Choiwonwong/demo-flink-kafka-redis-api

다른 공부 하려다 한번은 올려봐야지 싶어, 급하게 구성하게 되었는데 Claude & Gemini의 강력한 도움을 받아 구성할 수 있었다. Shell 스크립트도 알아서 구성해줘서, 혼자 했으면 2일은 꼬박 걸렸을 거 같은데 2~3시간 정도 만에 마무리가 가능했다.

Flink 애플리케이션은 직접 구현하기 위해선 별도로 많은 시간을 투자해야할 것 같았다.
핵심 설정은 알더라도, 동작하게 만들기 위해선 집계함수도 만들고, 클래스도 정의하고 여러가지 작업이 들어가야해서 정말 간단하게 올리는 케이스에선 Kafka Steams이 좋지 않을까 생각된다.

알고보니 Flink LST 버전이 1.20이던데, 1.18.0 버전으로 구성된걸 보아 Claude는 1.18.0 버전까지 알고있는 듯 하다.

언젠가 이런 기술들도 실제 업무에서 적용할 수 있길 바라며, 포스팅을 마무리한다.

Reference

https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/operators/windows/
https://nightlies.apache.org/flink/flink-docs-master/docs/concepts/time/#notions-of-time-event-time-and-processing-time

728x90

'Tech > Data Engineering' 카테고리의 다른 글

Spark Join Strategy 로컬에서 확인해보기 (1)	2025.09.10
Spark BufferHolder 메모리 한계 해결하기 - explode -> RDD Flatmap, 제너레이터 (3)	2025.08.19
Hive Json Serde의 한계 - Parquet/ORC 포맷 (2)	2025.08.18
Redis를 활용한 데이터 수집 프로세스 병렬 처리 및 안정성 확보 (5)	2025.08.17
Hive Table 복구 회고 with Airflow catchup 설정 (0)	2024.07.27