Kafka와 Spark Structured Streaming으로 설계하는 실시간 분석 파이프라인

이벤트가 들어온 뒤 분석 결과가 나올 때까지

실시간 분석은 데이터를 빠르게 받는 일만으로 완성되지 않는다. 앱·IoT·로그에서 들어오는 이벤트의 스키마를 통제하고, 늦게 도착하거나 중복된 데이터를 처리하며, 장애 뒤에도 일관된 결과를 복구해야 한다.

Kafka는 대규모 이벤트를 파티션 기반 로그에 영구 저장하고, 소비자가 오프셋으로 처리 위치를 관리하는 분산 스트리밍 플랫폼이다. Idempotent Producer와 Producer/Consumer 트랜잭션은 처리 일관성을 위한 기반이 된다.

Spark Structured Streaming은 DataFrame/Dataset 선언형 API를 통해 마이크로배치 또는 연속 처리를 수행한다. 이벤트 시간, 워터마크, 상태 연산을 사용해 윈도우·조인·집계 과정에서 지연 및 중복 이벤트를 제어한다.

이 둘을 실시간 OLAP이나 피처 스토어와 연결하면 스트리밍 ETL 결과를 즉시 질의하거나 온라인 추론에 공급할 수 있다.

수집·처리·서빙 계층을 나누어 설계한다

수집 계층의 Kafka는 파티션, 리플리케이션, ACK와 배치 튜닝을 통해 고가용성과 처리량을 맞춘다. Avro 또는 Protobuf 기반 스키마 레지스트리를 붙이면 스키마 진화와 호환성 관리도 가능하다.

처리 계층에서는 Spark Structured Streaming이 이벤트 시간 기준의 윈도우와 워터마크를 적용한다. 체크포인트와 WAL은 장애 복구에 쓰이며, 정확히 한 번 처리는 지원 싱크 범위 안에서 보장된다.

서빙 계층은 용도에 따라 나뉜다. ClickHouse, Druid, Apache Pinot 같은 실시간 OLAP은 서브초 쿼리 지연을 목표로 하고, Delta Lake·Hudi·Iceberg는 스트리밍과 배치를 함께 다루면서 머저블 히스토리를 유지한다.

운영 계층에는 속도·지연·오프셋 랙 메트릭, 로그, 트레이싱을 연결한다. ACL, TLS, SASL, 데이터 마스킹, PII 토큰화도 데이터 흐름과 분리하지 않고 함께 설계해야 한다.

파이프라인에서 실패를 처리하는 경로

입력 이벤트는 스키마 레지스트리와 연결된 Kafka 토픽으로 유입된다. Spark는 이를 파싱·정제하고, 윈도우 집계와 상태 관리를 거쳐 OLAP·캐시·피처 스토어로 전달한다. 파싱 오류나 스키마 불일치는 DLQ로 분기하며, 이후 재시도·DLQ·오프라인 백필의 경로로 교정한다.

체크포인트와 트랜잭셔널 Kafka 싱크는 정확히 한 번 처리 설계에 사용된다.

⤢✕

구성 요소별 운영 특성

구성 요소	성능(지연/처리량)	확장성	일관성	안정성	운영 편의
Kafka	ms~수십 ms, GB/s+	파티션 선형 확장	트랜잭션/Idempotent	브로커 RF로 고가용	중간, 토픽/파티션 관리 필요
Spark Structured Streaming	수백 ms~수초(마이크로배치)	클러스터 수평 확장	체크포인트 기반 EOS(지원 싱크)	장애 시 재처리 복구	중간, 스테이트/셔플 튜닝 필요
실시간 OLAP(Druid/ClickHouse/Pinot)	50~500 ms 쿼리	샤딩·리플리카 확장	보통 Eventually Consistent	노드 분산 복구	중간~상, 인덱스/롤업 설계 필요

구현 전에 맞춰둘 환경과 계약

Kafka 3.6+, Zookeeper 또는 KRaft, JDK 11이 필요하다. Spark는 3.5.x와 Python 3.10 또는 Scala 2.12 조합을 사용하며, kafka-python 또는 confluent-kafka, PySpark를 준비한다. 네트워크, TLS, ACL 구성도 선행 조건이다.

토픽 설계에서는 고카디널리티 키를 분산하고 핫파티션을 피할 수 있도록 키 파티셔닝 전략을 정한다. retention.ms와 segment.bytes를 조정하고, 스키마 레지스트리에서 Backward 또는 Full 호환성 모드를 설정한다.

프로듀서와 컨슈머의 기준점은 배치 전송, 압축, ACK 설정의 균형이다. linger.ms, batch.size, zstd/snappy 압축, ACK(all)을 함께 검토하고 Idempotent producer=true, enable.idempotence를 적용한다.

스트리밍 잡은 입력, 정제, 스키마 적용, 윈도우·워터마크, 집계·조인, 서빙 싱크, 체크포인트 순서로 구성한다. 상태 TTL과 메모리·디스크 스토어 용량은 워터마크 정책과 함께 관리한다.

Kafka 이벤트를 보내는 프로듀서

전제: pip install kafka-python, Kafka 브로커 PLAINTEXT on localhost:9092

# python3 -m pip install kafka-python
from kafka import KafkaProducer
import json, time, random

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    acks='all',
    linger_ms=20,
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    key_serializer=lambda k: k.encode('utf-8'),
    retries=5,
    compression_type='zstd',
    enable_idempotence=True
)

topic = "events.clicks"
while True:
    evt = {
        "event_time": int(time.time() * 1000),
        "user_id": f"u{random.randint(1, 100000)}",
        "page": random.choice(["home","prod","cart"]),
        "cost": round(random.random() * 10, 2)
    }
    key = evt["user_id"]
    producer.send(topic, key=key, value=evt)
    producer.flush()
    time.sleep(0.05)

Spark에서 윈도우 집계를 실행하는 잡

전제: spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0

# spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0 stream.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, window, sum as s, expr
from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType

spark = SparkSession.builder.appName("rt-analytics").getOrCreate()
spark.conf.set("spark.sql.shuffle.partitions", "200")

schema = StructType([
    StructField("event_time", LongType(), False),
    StructField("user_id", StringType(), False),
    StructField("page", StringType(), False),
    StructField("cost", DoubleType(), True)
])

df = (spark.readStream.format("kafka")
      .option("kafka.bootstrap.servers","localhost:9092")
      .option("subscribe","events.clicks")
      .option("startingOffsets","latest")
      .load())

parsed = (df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
            .select(from_json(col("value"), schema).alias("v"))
            .selectExpr(
                "timestamp_millis(v.event_time) AS event_ts",
                "v.user_id","v.page","v.cost"))

agg = (parsed
       .withWatermark("event_ts","5 minutes")
       .groupBy(window(col("event_ts"), "1 minute"), col("page"))
       .agg(s("cost").alias("revenue"), expr("count(1) as cnt")))

# 출력 1: 콘솔(디버그)
query1 = (agg.writeStream
          .outputMode("append")
          .format("console")
          .option("truncate","false")
          .option("numRows","20")
          .option("checkpointLocation","/tmp/ckpt/console")
          .start())

# 출력 2: Kafka 싱크(트랜잭셔널 EOS)
out = (agg.selectExpr(
        "CAST(page AS STRING) AS key",
        "to_json(named_struct('window_start', window.start, 'window_end', window.end, 'revenue', revenue, 'cnt', cnt)) AS value"))

query2 = (out.writeStream
          .format("kafka")
          .option("kafka.bootstrap.servers","localhost:9092")
          .option("topic","rt.metrics")
          .option("kafka.transactional.id","spark-tx-rt-analytics")
          .option("checkpointLocation","/tmp/ckpt/kafka")
          .outputMode("append")
          .start())

spark.streams.awaitAnyTermination()

이 구성은 watermark 5분으로 지연 데이터를 허용하면서 1분 윈도우 집계를 수행한다. kafka.transactional.id는 Kafka 싱크에 정확히 한 번 쓰기를 보장하고, 싱크별 체크포인트 경로 분리는 독립적인 복구를 가능하게 한다.

실시간 반영이 필요한 업무

카드 결제나 로그인 이벤트에서는 윈도우 기반 이상치 스코어링 결과를 Feature Store에 동기화해 온라인 모델 추론으로 보낼 수 있다.

개인화와 추천에서는 클릭스트림 집계 결과를 신호 업데이트와 실시간 랭킹에 연결한다. 사용자 세션은 지연 1~3초 내에 반영한다.

SRE 로그 분석은 에러율과 지연 P95 지표를 스트리밍으로 집계해 알럿을 발송하고, 파싱 실패 로그는 DLQ에 모아 백필한다. IoT와 제조 현장에서는 센서 데이터의 윈도우 통계와 스펙 외 편차를 탐지해 현장 대시보드에 서브초 응답으로 제공한다.

성능과 신뢰성을 함께 다루는 운영 기준

Kafka 파티션 수는 피크 처리량과 컨슈머 병렬성을 기준으로 산정한다. 파티션당 5~15MB/s 가이드를 두고 linger.ms, batch.size, 압축 설정으로 네트워크 효율을 조정한다. Spark는 maxOffsetsPerTrigger로 소스 인입량을 제어하고, state TTL과 워터마크로 상태 폭증을 막는다. Kryo 직렬화와 적절한 shuffle 파티션 수도 함께 조정한다.

OLAP에서는 롤업, 세그먼트 크기, 인덱스를 설계해 브루트포스 탐색을 피한다. 정확히 한 번 처리를 위해 체크포인트와 트랜잭셔널 싱크를 적용하고, 멱등 키를 설계한다. DLQ에 축적된 이벤트는 백필 잡으로 교정하며, 오프셋 재설정 절차는 표준화한다.

전송·저장 암호화(TLS, at-rest), SASL/OAUTH, RBAC/ACL 최소 권한, PII 마스킹과 토큰화, 스키마 레벨 정책 태깅도 운영 범위에 포함된다. 보안 계층을 추가하면 지연이 증가하는 트레이드오프가 생긴다.

Kafka 저장 용량은 (일일 인입량 × 보존일 × 복제계수)로 계산한다. 예를 들어 2TB/일 × 7일 × RF3 ≈ 42TB 원시다. 스토리지·네트워크·컴퓨트를 스케일아웃할 수 있도록 준비하고, 계측 결과를 바탕으로 오토스케일한다.

탐지와 의사결정 지연은 10분에서 1분 이하로 단축하고, 알람 MTTR은 30% 이상 줄일 수 있다. 배치와 스트림을 통합하면 중복 파이프라인을 제거해 머지·보관 최적화 가정 아래 인프라 비용을 15~25% 절감할 수 있다. 데이터 신선도는 사용자 경험에 영향을 주며, 스키마 진화와 거버넌스 자동화는 변경 대응의 탄력성을 높인다.