Kafka

 

Kafka란 무엇인가?

ㆍ분산 메시징 시스템으로 대용량의 실시간 로그처리에 특화된 아키텍처 설계를 통하여 기존 메시징 시스템보다 우수한 TPS (Transaction Processing System)를 제공함

 

기본 구성과 동작

• producer : topic 메세지를 생성한 뒤 해당 메세지를 broker에 전달
• broker : 전달받은 메세지를 topic 별로 쌓아놓음
• cluster(Zookeeper) : kafka는 확장성(scale-out)과 고가용성(high availability)을 위하여 broker들이 클러스터로 구성되어 동작하도록 설계되어있다. 심지어 broker가 1개 밖에 없을 때에도 클러스터로써 동작한다.
• consumer : 해당 topic을 pull로 메시지를 가져와서 처리

※ 클러스터란 ? 여러 대의 컴퓨터들이 연결되어 하나의 시스템처럼 동작하는 컴퓨터들의 집합

자료 출처 : http://epicdevs.com/17

 

특징

• 분산 시스템을 기본으로 설계되었기 때문에, 기존 메시징 시스템에 비해 분산 및 복제 구성을 손쉽게 할 수 있다.
• 단순한 메시지 헤더를 지닌 TCP기반의 프로토콜을 사용하여 프로토콜에 의한 오버헤드를 감소시킨다.
• 다수의 메시지를 batch형태로 broker에게 한 번에 전달할 수 있어 TCP/IP 라운드트립 횟수를 줄일 수 있다.
• 메시지를 메모리가 아닌 파일 시스템에 저장한다. 영속성(durability)이 보장된다.
• cosumer가 push 방식이 아닌 pull 방식으로 동작한다.

 

Topic과 Partition

• Kafka의 topic은 partition이라는 단위로 쪼개어져 클러스터의 각 서버들에 분산되어 저장되고, 고가용성을 위하여 복제(replication) 설정을 할 경우 이 또한 partition 단위로 각 서버들에 분산되어 복제되고 장애가 발생하면 partition 단위로 fail over가 수행된다.

 

Consumer와 Consumer Group

• 메시징 모델은 크게 큐(queue) 모델과 발행-구독(publish-subscribe) 모델로 나뉜다. 큐 모델은 메시지가 쌓여있는 큐로부터 메시지를 가져와서 consumer pool에 있는 consumer 중 하나에 메시지를 할당하는 방식이고, 발행-구독 모델은 topic을 구독하는 모든 consumer에게 메시지를 브로드캐스팅하는 방식이다. Consumer group을 구성하는 consumer의 수가 partition의 수보다 작으면 하나의 consumer가 여러 개의 partition을 소유하게 되고, 반대로 consumer의 수가 partition의 수보다 많으면 여분의 consumer는 메시지를 처리하지 않게되므로 partition 개수와 consumer 수의 적절한 설정이 필요하다.

샘플 구축

• 다운로드 링크 : https://www.apache.org/dyn/closer.cgi?path=/kafka/1.1.0/kafka_2.11-1.1.0.tgz
• zookeeper 서버 실행

bin/zookeeper-server-start.sh config/zookeeper.properties

• KafKa Borker 서버 실행

bin/kafka-server-start.sh config/server.properties

• Topic 생성

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic request
bin/kafka-topics.sh --list --zookeeper localhost:2181

• Producer에서 메세지 전송

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic request

• Cosumer에서 메세지 pull로 가져오기

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic request --from-beginning

 

Python 코드로 샘플 구축

• consumer 실행

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    bootstrap_servers='localhost:9092',
)

consumer.subscribe(['request'])
print('[started]')
for msg in consumer:
    print(msg)
print('[end]')

• producer 실행하여 임의의 메세지 전송

import json
from kafka import KafkaProducer

KAFKA_BROKERS = (
    'localhost:9092',
)

producer = KafkaProducer(bootstrap_servers=KAFKA_BROKERS, value_serializer=lambda v: json.dumps(v).encode('utf-8'))
producer.send('request', {'key': 'Hello Python Kafka!!'})
producer.flush()

• 전송 받은 메세지 확인

[started] ConsumerRecord(topic=u'request', partition=0, offset=4, timestamp=1534901810250, timestamp_type=0, key=None, value='{"key": "Hello Python Kafka!!"}', checksum=None, serialized_key_size=-1, serializ ed_value_size=31)