Apache Kafka 실무 스터디 가이드

1. Kafka란? — 왜 써야 하는가

정의

Apache Kafka는 분산 이벤트 스트리밍 플랫폼이다. LinkedIn에서 처음 개발되었고, 현재 Apache Software Foundation이 관리한다. 단순 메시지 큐가 아니라 실시간 데이터 파이프라인의 중추 신경 역할을 한다.

3대 핵심 능력

왜 Kafka를 쓰는가? (vs 전통 MQ)

대표 유즈케이스

• 실시간 로그 수집(Log Aggregation) — 여러 서버의 로그를 중앙으로 모은다
• 이벤트 소싱(Event Sourcing) — 상태 변경을 이벤트로 기록
• CDC(Change Data Capture) — DB 변경분을 실시간 전파 (Debezium)
• 실시간 스트림 처리 — Flink/Spark Streaming과 연동
• 마이크로서비스 간 통신 — 비동기 이벤트 기반 아키텍처
• ML Feature Store 업데이트 — 실시간 피처 파이프라인

2. 핵심 개념 6가지

토픽-파티션-오프셋 관계도

Consumer Group 동작 방식

Replication (복제)

각 파티션은 여러 브로커에 복제된다. 하나가 Leader(읽기/쓰기 담당), 나머지는 Follower(복제만). Leader 장애 시 Follower 중 하나가 새 Leader로 승격된다.

• replication.factor=3 → 데이터 3벌 보관 (운영 권장값)
• ISR (In-Sync Replicas) — Leader와 동기화된 Replica 집합. ISR에 포함된 Follower만 Leader 후보가 될 수 있다.

3. 아키텍처 한눈에 보기

2계층 구조

5대 핵심 API

4. KRaft 모드 — ZooKeeper 없는 Kafka

KRaft란?

KRaft(Kafka Raft)는 ZooKeeper를 대체하는 Kafka 자체 내장 메타데이터 관리 프로토콜이다. Raft 합의 알고리즘의 이벤트 기반 변형을 사용하여, 별도 클러스터(ZooKeeper) 없이 Kafka 브로커만으로 메타데이터를 관리한다.

KRaft의 장점

• 운영 단순화 — ZooKeeper 클러스터를 별도로 관리할 필요 없음
• 더 빠른 컨트롤러 페일오버 — 메타데이터가 Kafka 내부에 있으므로 전환 속도 향상
• 확장성 향상 — 수백만 파티션까지 더 효율적으로 관리
• 배포 간소화 — 하나의 시스템만 관리하면 됨

KRaft 모드 설치 (로컬 개발용)

# 1. 클러스터 UUID 생성
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

# 2. 로그 디렉토리 포맷
bin/kafka-storage.sh format \
  -t $KAFKA_CLUSTER_ID \
  -c config/kraft/server.properties

# 3. Kafka 서버 시작 (KRaft 모드)
bin/kafka-server-start.sh config/kraft/server.properties

KRaft 핵심 설정

# config/kraft/server.properties 주요 항목

# 노드 역할: controller, broker, 또는 둘 다 (개발용: controller,broker)
process.roles=broker,controller

# 노드 ID (클러스터 내 고유)
node.id=1

# 컨트롤러 쿼럼 투표자 목록
controller.quorum.voters=1@localhost:9093

# 리스너 설정
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
controller.listener.names=CONTROLLER

# 로그 디렉토리
log.dirs=/tmp/kraft-combined-logs

5. 5분 퀵스타트 (로컬 설치)

Step 1: 다운로드 & 설치

# Kafka 최신 버전 다운로드 (https://kafka.apache.org/downloads)
wget https://downloads.apache.org/kafka/4.0.0/kafka_2.13-4.0.0.tgz
tar -xzf kafka_2.13-4.0.0.tgz
cd kafka_2.13-4.0.0

Step 2: KRaft 모드로 시작

# 클러스터 ID 생성 & 스토리지 포맷
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID \
  -c config/kraft/server.properties

# Kafka 서버 시작
bin/kafka-server-start.sh config/kraft/server.properties

Step 3: 토픽 생성

# 토픽 생성 (파티션 3개)
bin/kafka-topics.sh --create \
  --topic my-first-topic \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1

# 토픽 목록 확인
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# 토픽 상세 정보
bin/kafka-topics.sh --describe \
  --topic my-first-topic \
  --bootstrap-server localhost:9092

Step 4: 메시지 생산 & 소비

# 터미널 1: Producer (메시지 입력)
bin/kafka-console-producer.sh \
  --topic my-first-topic \
  --bootstrap-server localhost:9092
> Hello Kafka!
> This is my first event.

# 터미널 2: Consumer (메시지 읽기)
bin/kafka-console-consumer.sh \
  --topic my-first-topic \
  --from-beginning \
  --bootstrap-server localhost:9092

6. Python 프로듀서 / 컨슈머 실습

라이브러리 선택 가이드

설치

# confluent-kafka (권장)
pip install confluent-kafka

# 또는 kafka-python
pip install kafka-python

Producer 예제 (confluent-kafka)

from confluent_kafka import Producer
import json

# 설정
conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'my-python-producer',
    # 성능 튜닝
    'linger.ms': 10,          # 배치 대기 시간
    'batch.size': 16384,      # 배치 크기 (bytes)
    'compression.type': 'lz4' # 압축
}

producer = Producer(conf)

# 전송 콜백
def delivery_callback(err, msg):
    if err:
        print(f'전송 실패: {err}')
    else:
        print(f'전송 성공: {msg.topic()} [{msg.partition()}] @ {msg.offset()}')

# 메시지 전송
for i in range(10):
    event = {'user_id': i, 'action': 'page_view', 'page': f'/products/{i}'}
    producer.produce(
        topic='user-events',
        key=str(i),               # 같은 key → 같은 파티션
        value=json.dumps(event),
        callback=delivery_callback
    )

# 버퍼에 남은 메시지 전송 완료 대기
producer.flush()

Consumer 예제 (confluent-kafka)

from confluent_kafka import Consumer, KafkaError
import json

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-consumer-group',
    'auto.offset.reset': 'earliest',   # 처음부터 읽기
    'enable.auto.commit': True,         # 자동 오프셋 커밋
    'auto.commit.interval.ms': 5000
}

consumer = Consumer(conf)
consumer.subscribe(['user-events'])

try:
    while True:
        msg = consumer.poll(timeout=1.0)  # 1초 대기

        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                print(f'파티션 끝 도달: {msg.topic()} [{msg.partition()}]')
            else:
                print(f'에러: {msg.error()}')
            continue

        # 메시지 처리
        event = json.loads(msg.value().decode('utf-8'))
        print(f'수신: partition={msg.partition()}, '
              f'offset={msg.offset()}, event={event}')

except KeyboardInterrupt:
    pass
finally:
    consumer.close()  # Consumer Group 탈퇴 & 오프셋 커밋

7. 핵심 설정값 레퍼런스

Broker 설정

Producer 설정

Consumer 설정

8. 메시지 전달 보장 (Delivery Semantics)

운영 권장 조합

# Producer: 안전한 기본값
acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5
retries=2147483647

# Broker: 데이터 안전성
min.insync.replicas=2
unclean.leader.election.enable=false

# Consumer: 수동 커밋으로 정확한 처리
enable.auto.commit=false
# → 메시지 처리 완료 후 consumer.commit() 호출

9. Kafka + AI/ML 파이프라인

왜 AI 파이프라인에 Kafka가 필요한가?

대부분의 AI 실패는 모델 정확도 문제가 아니라 데이터 파이프라인 병목에서 온다. 배치 파이프라인으로 실시간 예측을 서빙하려는 것이 근본 원인이다. Kafka는 이 문제를 해결하는 실시간 데이터 백본이다.

핵심 아키텍처 패턴

실무 시나리오별 적용

Kafka-ML 프레임워크

Kafka-ML은 Kafka의 고처리량·저지연 특성을 활용하여 ML/AI 파이프라인을 관리하는 오픈소스 프레임워크다. TensorFlow와 PyTorch를 지원하며, 연속적인 데이터 스트림에서 동작하는 ML 시스템 구축에 특화되어 있다.

모델 드리프트 대응

Kafka + Flink 조합으로 온라인 ML 학습과 모델 드리프트 감지를 구현할 수 있다. 연속적인 모델 훈련과 실시간 추론을 통해 항상 최신 상태의 고정확도 예측을 유지할 수 있다.

간단한 예제: Kafka → ML 추론

from confluent_kafka import Consumer, Producer
import json, pickle

# 학습된 모델 로드
model = pickle.load(open('fraud_model.pkl', 'rb'))

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'ml-inference',
    'auto.offset.reset': 'latest'
})
consumer.subscribe(['transactions'])

producer = Producer({'bootstrap.servers': 'localhost:9092'})

while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue

    # 1. 이벤트 파싱
    event = json.loads(msg.value().decode('utf-8'))

    # 2. 피처 추출
    features = extract_features(event)  # 사용자 정의 함수

    # 3. 모델 추론
    prediction = model.predict([features])[0]
    score = model.predict_proba([features])[0][1]

    # 4. 결과를 다른 토픽에 발행
    result = {
        'transaction_id': event['id'],
        'is_fraud': bool(prediction),
        'fraud_score': float(score)
    }
    producer.produce(
        'fraud-results',
        key=str(event['id']),
        value=json.dumps(result)
    )
    producer.flush()

10. 실무 체크리스트 & 트러블슈팅

프로덕션 배포 전 체크리스트

자주 겪는 문제 & 해결

Consumer Lag 증가

• 원인: Consumer 처리 속도 < Producer 속도
• 해결: 파티션 수 늘리기 + Consumer 인스턴스 추가, max.poll.records 조정, 처리 로직 최적화

Rebalancing이 너무 자주 발생

• 원인: session.timeout.ms가 너무 짧거나, max.poll.interval.ms 초과
• 해결: session.timeout.ms 늘리기 (45s → 60s), 무거운 처리는 별도 스레드로 분리

메시지 순서가 뒤바뀜

• 원인: 같은 key의 메시지가 다른 파티션에 들어감, 또는 재시도 시 순서 역전
• 해결: key 기반 파티셔닝 확인, enable.idempotence=true + max.in.flight.requests.per.connection=5

디스크 부족

• 원인: retention 기간이 너무 길거나, 토픽이 과도하게 많음
• 해결: log.retention.hours 줄이기, log.retention.bytes 설정, 불필요한 토픽 정리

모니터링 필수 메트릭

참고 자료

• Apache Kafka 공식 소개
• Kafka Quickstart Guide
• Kafka 공식 문서 (전체)
• Confluent - Kafka Architecture Course
• Confluent - What is Apache Kafka?
• Confluent - ML in Kafka Applications
• Kafka-ML GitHub (오픈소스 ML 파이프라인)
• Baeldung - ZooKeeper to KRaft
• kafka-python Documentation
• Confluent Python Client