[Apache] Apache Kafka의 개념

Apache Kafka란?

• 실시간 데이터 스트리밍 플랫폼
• 데이터를 발행(Publish)하는 쪽과 구독(Subscribe)하는 쪽을 비동기적으로 연결해주는 메시지 큐 시스템

-> 대용량 데이터를 빠르고 안정적으로 주고받는 실시간 데이터 파이프라인

 

 

기본 개념 구조

 

[Producer] → [Kafka Broker] → [Consumer]

 

• Producer : 데이터를 발행(전송)하는 주체 (ex : 애플리케이션, 센서, 서버 로그 등)
• Consumer : 데이터를 구독(수신)하는 주체 (ex : Spark, ELK, DB 등)
• Broker : Kafka 서버로 실제 데이터를 저장하고 전달
• Topic : 메시지를 분류하는 논리적 채널
• Partition : Topic을 분할한 단위
• Offset : 각 메시지가 Partition 내에서 차지하는 고유한 위치
• Consumer Group : 여러 Consumer가 하나의 Topic을 병렬로 처리하기 위한 그룹 단위

 

동작 흐름 예시

• 예시로 쇼핑몰 로그를 실시간 분석한다고 가정

[웹 서버 로그] → [Kafka Topic: web_log] → [Spark Streaming or Elasticsearch]

 

1. Producer (웹서버)
   • 사용자의 클릭, 결제 등의 로그를 JSON 형태로 Kafka에 전송
2. Kafka Broker
   • 로그를 'web_log' 라는 Topic에 저장
   • 여러 Broker가 있으면 데이터를 Partition으로 나눠 분산 저장
3. Consumer (Spark, Flink, ELK 등)
   • 'web_log' Topic을 구독하여 데이터를 읽음
   • 실시간으로 집계, 필터링, 알림 등 처리 수행

 

Kafka 주요 특징

• 초고속 처리 : 초당 수백만 건의 메시지를 처리 가능
• 분산 구조 : 여러 Broker로 구성되어 장애에 강함
• 확장성 : Topic/Partiiton을 추가하여 선형 확장
• 내구성 : 데이터를 디스크 형태로 저장 -> 장애 후 복구 가능
• 실시간 스트리밍
• Exactly-once 보장 : 동일 메시지가 중복 처리되지 않도록 제어 가능

 

메시지 처리 모델

• Kafka의 기본 원칙은 Pub/Sub 모델
• 각 메시지는 순서가 있고 Consumer는 각 Partition에서 Offset을 기억하여 어디까지 읽었는지 관리
  • 즉 Consumer가 죽어도 복구 후 다시 그 Offset부터 이어서 읽을 수 있음

 

Java 예시 코드

// Producer 예시
import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 10; i++) {
            producer.send(new ProducerRecord<>("test-topic", "key" + i, "message-" + i));
        }

        producer.close();
    }
}

// Consumer 예시
import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "group1");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset=%d, key=%s, value=%s%n",
                        record.offset(), record.key(), record.value());
            }
        }
    }
}

 

 

Kafka의 활용

• 실시간 로그 수집
• 금융 거래 처리 (ex : 실시간 결제, 이상 거래 탐지)
• IoT 데이터 스트림 (ex : 센서, 스마트 디바이스 데이터 수집)
• 데이터 파이프라인 허브 (다양한 시스템 간 실시간 데이터 전송)
• 머신러닝 피드백 루프 (모델 예측 -> 사용자 반응 -> 모델 재학습)