Kafka

---

 

실무에서 사용하던 개념들 정리..!

 

---

1. Apache Kafka?

 

Apache Kafka 개요

• 실시간 기록 스트림을 게시/구독/저장 및 처리할 수 있는 데이터 분산 스트리밍 플랫폼 
• 대용량 실시간 로그처리에 특화되어 있는 솔루션
• Fault-Tolerant : 안정적인 퍼포먼스
• 범용메시징 시스템 : 메세지를 파일시스템에 저장
• 단순한 메세지 헤더를 지닌 TCP기반의 프로토콜을 사용(프로토콜로 인한 오버헤드 감소)

 

Apache Kafka 특징

• Publisher - Subscriber Model : 데이터 큐를 중간에 두고 서로 간 독립적으로 데이터를 생산하고 소비
  의존성이 없으르모 publishers나 subscriber가 죽어도 안정적으로 데이터를 처리할 수 있음
• 고가용성 및 확장성 : 클러스터로서 작동하므로 Fault-tolerant한 고가용성 서비스를 제공할 수 있고 
  빠른 데이터 처리를 가능하게 함. 서버를 수평적으로 늘려 안정성 및 성능을 향상시키는 Scale-out이 가능
•  디스크 순차 저장 및 처리 : 메세지를 디스크에 순차적으로 저장
    - 서버에 장애가 발생해도 디스크 유실 걱정이 없음
    - 순차적을 저장되므로 디스크 I/O가 줄어 성능이 빨라짐
• 분산처리 : 여러개의 파티션을 서버들에 분산시켜 나누어 처리, 메세지를 상황에 맞추어 빠르게 처리할 수 있음

 

Apache Kafka 사용할 경우

• 다른 곳의 API로 부터 데이터 송수신, 다양한 Application에서 비동기 통신
• 실시간으로 처리 할 용량이 많이 꽤 클 것으로 예상되는 경우 (실시간보다 배치로 처리할 양이 많을 경우)
• 클라우드 기반에서 사용할 때 무료크레딧을 써서 작은 공간이나 작은 메모리 여러개를 사용할 경우  (많은 양의 process를 처리 , email 발송 및 upload 가능)
• 프로젝트가 버전업 될 때마다 멀티 아웃으로 추가 할 기술스택이 많아질 경우 (확장성을 고려해야 할 경우)

 

Kafka APIs

• Admin API : topic, broker, 다른 kafka 객체 관리
• producer API
• consumer API
• Kafka Streams API : event-time 에 근거한 aggregations, joins, windowing, processing
• Connect API : 재사용 가능한 data를 import / export 가능토록 하는 Connector 역할 (read or write streams of events)

 

---

2. Message Q  (MQ)

 

Message Q 개요

• Proccess / Program instance 가 데이터를 서로 교환할 때 사용하는 통신방법
• 메세지 지향 미들웨어 (MOM) : 비동기 메세지를 사용하는 응용 프로그램 간의 데이터 송수신

 

Message Q 특징

• Asynchronous : Queue 에 넣기 때문에 비동기적으로 데이터 처리 ( 송신측에 관계없이 수신측에서 수신호로 타이밍을 식별 )
• Decoupling  : application과 분리가능
• Resilience : 일부 실패시 전체에 영향을 주지 않음
• Redundancy : 실패시 재실행 가능
• Guarantees : 처리된 작업 확인 가능 (보증)
• Scalable : 다수의 프로세스들이 큐에 메세지를 보낼 수 있음

 

Message Q 종류

• RadditMQ : AMQT 프로토콜을 구현해놓은 프로그램, 신뢰성, 유연한 라우팅, 관리 UI의 편리성 (빠르고 쉬우며 직관적)
• Kafka : (Camel에서 client로 사용) : 확장성, 고성능 및 높은 처리량
  - 대용량 실시간 로그 처리에 특화 (단순한 메시지 헤더를 지닌 TCP 기반의 프로토콜 사용으로 오버헤드 감소)
  - 분산 시스템으로 인해 분산 및 복제 구성 장점 (노드 장애에 대한 대응성을 가지고 있음)
  - 프로듀서는 각 메시지를 배치로 broker에게 전달하여 TCP/IP 라운드 트립을 줄임
  - 기본적으로는 파일시스템에 저장을 통해 영속성(수명)을 보장 (오류시 오류 지점부터 복구가 가능)
  - Kafka, Kinesis(aws)는 window 단위의 데이터를 넣고 꺼낼 수 있다
• ActiveMQ (Camel에서 client로 사용) : 효율적이고 사용하기 쉬움
  - Java로 만든 OpenSource Message Broker
  - 다양한 언어와 Protocol 지원 (Java, C, C++, Python, PHP Client 등)
  
  
  

Kafka vs RabbitMQ vs ActiveMQ : 업무에 따라 목적이 다름

• Kafka(Apache) : 처리량이 많은 분산 messaging system
• RadditMQ, ActiveMQ(Apache) : Kafka에 비해 신뢰성이 좋은 message broker (상대적)
  
  

---

3. Kafka Archetecture 

• Cluster를 중심으로 Producer 와 Consumer 가 데이터를 Push-Pull 하는 구조

 

 

 

Producer  : 데이터를 발생시키고 Kafka cluster에 적재하는 process

Consumer Group : 컨슈머 그룹으로서 데이터 처리시 컨슈머 수만큼 파티션의 데이터를 분산처리

 

 

Kafka Cluster : Kafka Server 로 구성된 Cluster

• Broker : Kafka Server, Topic 기준으로 메세지 관리 (Cluster로 구성)
• Topic : Kafka Cluster 에 데이터 관리시 그 기준이 되는 개념, 데이터를 관리하는 하나의 그룹
• Partition : 각 토픽 당 데이터를 분산 처리하는 단위, 파티션을 나누어 그 수대로 데이터를 분산처리

(카프카 옵션에서 지정한 replica의 수만큼 파티션이 각 서버들에게 복제됨, 파티션들은 운영도중 수를 늘릴 수 있지만
줄일수 없음)

• Leader / Follower : Leader 는 모든 읽기/쓰기 연산을 담당, Leader를 제외한 나머지는 Follower가 되고 단순히 리더의 데이터를 복사하는 역할을 담당
• ISR (in-sync replica) : Leader partition 를 복제(쓰기)하여 구성된  Follower

(만약 Learder가 복제를 실패할 경우, Kafka는 새로운 ISR를 새로운 Leader로 선출함, 복제본이 Leader를 따라잡았지 여부를 확인)

 

 

 

Zookeeper 개요

• 분산 코디네이션 System, Kafka Cluster Leader를 선출
• Master-Slave 아키텍처 기반으로 Leader Folloer 구성
• 개발자가 코디네이션 로직이 아닌 비지니스 핵심로직에 지원하도록 도움
• 무분별한 쓰기 동작으로 인한 경쟁상태를 예방

 

 

Zookeeper 특징

• 설정관리: 클러스터 설정 정보를 최신으로 유지하기 위한 조율 시스템으로 사용
• 클러스터 관리: 클러스터의 서버가 추가되거나 제외될때 그 정보를 클러스터 안 서버들이 공유하는 데 사용
• 리더채택 : 다중 어플리케이션 중에 어떤 노드를 선출할지 정하는 로직을 만드는데 사용

       (주로 복제된 여러 노드 중 연산이 이루어지는 하나의 노드를 택하는데 사용)

• 락, 동기화 서비스: 클러스터에 쓰기 연산이 빈번할 경우 경쟁상태에 들어갈 가능성이 생기므로 (불일치 발생)
  클러스터 전체를 대상을 동기화해 (락상태) 경쟁상태에 들어가는 경우를 사전에 방지

 

---

4. Kafka install / Settings (window)

 

경로/ 환경변수를 잘 확인하자

 

① zookeeper install : http://www.apache.org/dyn/closer.cgi/zookeeper/ 

② zookepper 경로 : C:\apache-zookeeper-3.6.2-bin\apache-zookeeper-3.6.2-bin\bin\zkServer.cmd

③ kafka 경로 : C:\kafka_2.13-2.7.0\kafka_2.13-2.7.0\bin\windows\kafka-server-start.bat config\server.properties

④ cmd 실행 :시작 -> 검색 -> cmd검색 후 실행 ->zookeeper 다운로드 경로 ( cd {경로} ) 입력 -> zkServer.cmd 실행

    UI 실행 : Zookeeper 다운로드 경로에서 zkServer.cmd 클릭

 

 

---

5. Kafka Command

 

• Zookeeper 시작	:  ./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
• Kafka 시작	: ./bin/kafka-server-start.sh -daemon config/server.properties
• Producer 시작	: ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topicname
• Consumer 시작	: ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topicname --from-beginning
• Topic 생성	: ./bin/kafka-topics.sh --creatae --zookeeper localhost:2181 --replication-factor1 --partitions 1 --topicname
• Topic List 조회	: ./bin/kafka-topics.sh --list --zookeeper localhost:2181
• 카프카 오프셋 리셋하고 다시 받기	: kafka-consumer-groups.sh --bootstrap-server kafka-domainLport --group kafkalocaltestatin --topic topicname --reset-offsets --to-datetime 2018-12-12T00:00.000 --execute kafka-console-consumer.sh --bootstrap-server kafka-domain-port --topic topicname --consumer-property group.id=kafkalocaltestatin
• Topic partition 1 보기	: /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --partition 1 --topic numtest
•  Topic 삭제	: /usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic my_topic
• Server 종료	: bin/kafka-server-stop.sh config/server.properties   bin/zookeeper-server-stop.sh config/zookeeper.properties
• Consumer Group 확인	: /usr/local/kafka/bin/kafka-consumer-groups.sh  --bootstrap-server localhost:9092 --list
• Consumer status / Offset 확인	: /usr/local/kafka/bin/kafka-consumer-groups.sh  --bootstrap-server localhost:9092 --group sr --describe
• Consumer group 삭제	: /usr/local/kafka/bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --delete --group <group_name>
• Topic leader/follower 확인	: /usr/local/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --topic my_topic --describe
• Server log 확인	: cat /usr/local/bin/kafka/logs/server.log

 

참조

• kafka-console-consumer.sh 상세 :  https://datacadamia.com/dit/kafka/kafka-console-consumer
• kafka-consumer-offset-rests 상세 : https://www.letmecompile.com/kafka-consumer-offset-reset/