Kafka 정리

사용 방식

• 분산 큐잉 시스템으로 사용하는 방법
  • 다른 프로세스에서 작동 중인 백그라운드의 테스크 프로세서에 요청하기 위한 큐로 사용하는 경우 등
• 데이터 허브로 사용하는 방법
  • 어떤 서비스에 데이터 업데이트가 발생시 해당 데이터를 사용하는 다른 여러 서비스에 알리기 위한 허브로 사용하는 방법

 

시스템 구조

 

용어 정리

• Cluster
  • 카프카 브로커들이 모여서 하나의 논리적인 단위를 이루는 집합
  • 분산된 방식으로 데이터를 저장하고, 확장성, 내구성, 가용성을 보장
• Broker
  • 카프카 시스템의 가장 기본적인 구성 요소
  • 카프카 클러스터에서 실행되는 각 서버/서비스
  • 프로듀서와 소비자 모두의 연결 지점 역할
• Zookeeper
  • 클러스터 관리 문제를 담당
    • 클러스터에 브로커 추가 및 제거, 리더/컨트롤러 브로커 결정, 토픽 구성 유지 등
  • 현재 버전에서는 사용 종료
• Topic
  • 데이터 관리 유닛
  • 데이터베이스나 메시지 큐에서 발견되는 큐와 유사한 구조
  • 데이터가 작성되는 즉시 모든 브로커가 액세스 가능
• Partition
  • 토픽을 나누는 단위
  • 클러스터 내 하나 이상의 브로커에 분산되어 저장
  • 병렬처리를 가능하게 함
• Producer
  • 토픽에 이벤트 메시지를 쓰는 클라이언트 애플리케이션
• Consumer
  • 토픽에서 이벤트 메시지를 읽는 클라이언트 애플리케이션

 

동작 방식 (예정)

 

카프카 설치 (MAC)

자바 설치

1. https://www.oracle.com/kr/java/technologies/downloads/#java8-mac 접속
2. dmg 파일을 다운로드
3. pkg 파일 실행
4. 설치 확인 → java -version

Kafka 설치

1. https://www.apache.org/dyn/closer.cgi?path=/kafka/2.8.0/kafka_2.13-2.8.0.tgz 접속
2. tgz 파일 설치 후 압축 해제

Kafka 실행

• kafka config 파일: /config/server.properties

// 터미널에서 바로 실행
$ bin/kafka-server-start.sh config/server.properties

// 데몬 프로세스로 실행
$ bin/kafka-server-start.sh -daemon config/server.properties

 

실습

Kafka 실습 (CLI)

• 토픽 생성

# 토픽명 "kafka.test-topic"
$ bin/kafka-topics.sh --create --topic kafka.test-topic --bootstrap-server localhost:9092

"""
옵션
--topic : 토픽 이름 지정
--bootstrap-server : 연결할 카프카 서버(브로커)
--replication-factor : replica 개수
--partitions : 토픽 파티션 개수
"""

• 토픽 확인

$ bin/kafka-topics.sh --bootstrap-server localhost:9092 --list localhost:2181

• Producer 콘솔 실행 & 메시지 전송

$ bin/kafka-console-producer.sh --topic kafka.test-topic --bootstrap-server localhost:9092
> "메시지1 입력"
> "메시지2 입력"
> "메시지3 입력"

• Consumer 콘솔 실행 & 메시지 수신

$ bin/kafka-console-consumer.sh --topic new-topic --from-beginning --bootstrap-server localhost:9092
"메시지1 입력"
"메시지2 입력"
"메시지3 입력"

• 트러블 슈팅
  • 토픽 확인 시 TimeoutException 발생
  • Kafka 브로커가 모든 네트워크 인터페이스에서 포트 9092로 들어오는 연결을 수신하도록 설정 필요
  • config/server.properties 확인

  ############################# Socket Server Settings #############################
  
  # The address the socket server listens on. It will get the value returned from
  # java.net.InetAddress.getCanonicalHostName() if not configured.
  #   FORMAT:
  #     listeners = listener_name://host_name:port
  #   EXAMPLE:
  #     listeners = PLAINTEXT://your.host.name:9092
  # listeners=PLAINTEXT://:9092  <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<

  • 표시 부분 주석 해제

 

파이썬 실습 예제 코드

• Producer

from kafka import KafkaProducer
import json

bootstrap_servers = ['localhost:9092']

producer = KafkaProducer(
    bootstrap_servers=bootstrap_servers,
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

message = {
    'key': 'value'
}

producer.send('kafka.test-topic', value=message)

producer.close()

print("Message sent successfully")

• Consumer

from kafka import KafkaConsumer
from json import loads
import time
 
consumer = KafkaConsumer(
    'kafka.test-topic',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='test-group',
    value_deserializer=lambda x: loads(x.decode('utf-8')),
    consumer_timeout_ms=1000
)

while True:
	records = consumer.poll()
	
	if records:
		for partition, messages in records.items():
			for message in messages:
				print(f'Offset: {message.offset}, Key: {message.key}, Value: {message.value}')
	else:
		time.sleep(1)

 

참고할만한 자료

• 기본 개념

https://medium.com/@cobch7/kafka-producer-and-consumer-f1f6390994fc

https://medium.com/@cobch7/kafka-architecture-43333849e0f4

• 카프카 동작

https://engineering.linecorp.com/ko/blog/how-to-use-kafka-in-line-1

https://engineering.linecorp.com/ko/blog/how-to-use-kafka-in-line-2

https://engineering.linecorp.com/ko/blog/how-to-use-kafka-in-line-3

• Best Practice

https://sunrise-min.tistory.com/entry/Kafka-ThroughputLatency-Best-Practices-공식문서

https://www.getorchestra.io/guides/apache-kafka-message-format

https://medium.com/apache-kafka-from-zero-to-hero/apache-kafka-guide-3-producers-and-message-keys-a215e5fff75b

https://levelup.gitconnected.com/kafka-client-best-practices-9b0761503df1

https://medium.com/@byteblog/kafka-best-practices-high-level-9afcfb92645a

https://blog.team-joon.com/카프카-메시지에-스키마를-정의해-보자-apache-avro-7162e250ae69

https://www.confluent.io/blog/avro-kafka-data/

https://developer.confluent.io/courses/event-design/best-practices/