[ 기초 개념 ]리플리케이션 Replication토픽 생성 명령 중 replication-factor 2 이라는 옵션을 설정하면원본을 포함한 복제본이 총 2개가 있다는 뜻입니다.리플리케이션 팩터 수가 커지면 안정성은 높아지지만브로커 리소스를 많이 사용하게 됩니다. 파티션 Partition하나의 토픽이 한 번에 처리할 수 있는 처리량을 늘리기 위해토픽을 여러 개로 나눠 병렬 처리가 가능하게 만든 것을 파티션이라고 합니다.나뉜 파티션 수만큼 컨슈머를 연결할 수 있습니다. 파티션은 언제든 늘릴 수 있지만 줄일 수는 없으므로,컨슈머의 LAG ( 프로듀서 메세지 수 - 컨슈머 메세지 가져간 수)가 0이 되도록 설정하는 것이 좋습니다. 참조 - 파티션과 컨슈머 할당에 대한 글 세그먼트 Segment프로듀서 의해 브..

[ 데비지움 ( Debezium ) 이란? ]변경 데이터 캡처(Change Data Capture, CDC)로,데이터베이스의 변경 사항을 실시간으로 추적하고 이를 스트림 형태로 전달하는 오픈 소스 분산 플랫폼입니다.주로 CDC 이벤트를 Kafka로 topic으로 전달하는데 사용하며, Sink connector로 설정됩니다.다른 방법으로 적용할 수도 있습니다.데비지움 서버를 이용한 배포Java 라이브러리 데비지움 엔진Debezium connector는 데이터베이스의 변경 로그(binlog, WAL 등)를모니터링하여 실시간으로 변경 사항을 감지합니다.  [ 예제 with Docker ]고려사항 ( Consideration )Debezium 튜토리얼에서는 Docker를 활용하여 ZooKeeper, Kafka,..

Kafka 운영시에는 AWS EC2에 설치하거나 온프로미스 서버에 환경셋팅 한다. 최근에는 쿠버네티스를 통해 운영하기도하는데, 로컬에서 테스트 및 이것저것 Toy Project하기에는 도커로 설치하는게 가장 심플하다. 도커 이미지는 bitnami image를 사용한다. confluentinc image도 있지만 bitnami가 pull 횟수가 가장 많다. kafka는 zookeeper 이미지도 설치해야한다. KRaft라고 zookeeper없이 설치하는 방법도 있지만 대부분의 문서가 kafka-zookeeper 한 set으로 설명하고 있고 HBase와 같이 zookeeper와 함께 사용하고 있는 프로그램들이 있기에 익숙해지는 것이 좋다. [ 프로젝트 설정 ] 일단 프로젝트를 만든다. 이 글에선 Pychar..

[ Ubuntu ] * 사전설치 Python3 sudo apt-get upgrade # Pip설치 sudo apt-get install python3-pip #Kafka Lib 설치 sudo pip3 install kafka-python [ 테스트 코드 ] from kafka import KafkaProducer import time producer = KafkaProducer( bootstrap_servers=['172.31.11.11:9092', '172.31.22.222:9092', '172.31.33.33:9092'] ) for i in range(1,50): producer.send('Topic-Name', str.encode('kafka:-%d' % i) ) time.sleep(0.5)

[ 토픽삭제 안되는 경우 ] server.properties 파일의 delete.topic.enable=true임에도 토픽이 삭제되지 않는경우 1. 카프카 dir.log파일과 관련 주키퍼로그 파일 삭제 2. 카프카 브로커를 재시작한다. 참조 : https://stackoverflow.com/questions/23976670/when-how-does-a-topic-marked-for-deletion-get-finally-removed : https://stackoverflow.com/questions/44564606/how-can-i-remove-kafka-topics-marked-for-deletion

에러메세지 ] ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer) kafka.common.InconsistentClusterIdException: The Cluster ID lTeGi4hnRaKFOHhVMQnfEg doesn't match stored clusterId Some(eaVbu41vRfSi-XzD7DxMog) in meta.properties. The broker is trying to join the wrong cluster. Configured zookeeper.connect may be wrong. at kafka.server.KafkaServer.startup(KafkaSe..

* 카프카 PATH로 cd하여 실행함을 가정한다. [ 실행과 중지 ][ 카프카 실행 ]./bin/kafka-server-start.sh ./config/server.properties --daemon [ 카프카 중지 ]./bin/kafka-server-stop.sh  [ 토픽 ][ 토픽생성 ]# 카프카 서버3대 (broker)에 복제를 3번하는 토픽을 만든다../bin/kafka-topics.sh --create \--zookeeper 172.123.123.111:2181, 172.123.123.222:2181, 172.123.123.333:2181 \--replication-factor 3 --partitions 3 \--topic kafka-topic-name [ 토픽확인 ]./bin/kafka-to..

Kafka 매개변수는 표 : https://www.ibm.com/support/knowledgecenter/ko/SSPFMY_1.3.5/com.ibm.scala.doc/config/iwa_cnf_scldc_kfk_prp_exmpl_c.html $ vi /usr/local/kafka/config/server.properties [ Server Basics ] broker.id=1 : 브로커를 구분하기위한 ID [ Socket Server Settings ] listeners=PLAINTEXT://:9092 : 카프카 소켓서버가 수신하는 주소. 기본포트는 9092 num.network.threads=3 : 네트워크 요청을 처리하는 데 사용되는 스레드 수 num.io.threads=8 : 입력 및 출력 조작에..