> For the complete documentation index, see [llms.txt](https://ssosso.gitbook.io/study/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://ssosso.gitbook.io/study/messagequeue/kafka/3.md). # 3. 카프카 설치 ## 카프카 설치 카프카 배포판인 컨플루언트 플랫폼을 사용한 카프카 클러스터 구축 방법 소개 1. 서버를 1대만 사용하여 구성 2. 서버를 여러대 사용하여 구성 ### 카프카 클러스터 환경 구축하기 * 배포판 : 컨플루언트, 클라우데라, 호튼웍스 등 * 배포판별 기능차이 존재. 필요한거에 맞게 선택 ### 카프카 구축 * OS 설치 : Linux, Max 만 지원 * 일부 루트권한 필요 * 여러대 사용 -> `/etc/hosts 네트워크` 설정 -> 서버간 서로의 `호스트명`으로 이름 확인 * JDK : 1.8 지원 * 다운로드 (1.8-202버전 이하로, 퍼블리 무료버전) * * * 다운로드 파일 -> /tmp 에 위치시킴 * CentOS ```bash # JDK 설치 $ sudo rpm -ivh /tmp/jdk-8u202-linux-x64.rpm # 환경변수 -> /etc/profile.d/java.sh 생성 $ sudo vi /etc/profile.d/java.sh expert JAVA_HOME=/usr/java/default expert PATH=$PATH:$JAVA_HOME/bin # 반영 $ source /etc/profile.d/java.sh # 확인 $ echo $JAVA_HOME $ java -version ``` * 컨플루언트 플랫폼의 `YUM 리포지터리 등록` ```bash # 컨플루언트 제공 공개키 등록 $ sudo rpm --import https://packages.confluent.io/rpm/5.0/archive.key # 파일작성 $ sudo vi /etc/yum.repos.d/confluent.repo [Confluent.dist] name=Confluent repository (dist) baseurl=https://packages.confluent.io/rpm/5.0/7 gpgmeck=1 gpgkey=https://packages.confluent.io/rpm/5.0/archive.key enabled=1 [Confluent] name=Confluent repository baseurl=https://packages.confluent.io/rpm/5.0 gpgmeck=1 gpgkey=https://packages.confluent.io/rpm/5.0/archive.key enabled=1 # 기존캐시 삭제 $ yum clean all # yum 사용가능한 패키지 목록 확인 $ yum list | grep confluent ``` * 카프카 설치 컨플루언트 플랫폼 -> 잘게 구분된 여러 패키지 존재함. 그중 `confluent-platform-oss-2.11`은 OSS 버전 패키지 설치, 컨플루언트 문서 참고 ```bash # 카프카 실행을 위한 패키지 설치 $ sudo yum install confluent-platform-oss-2.11 ``` * 브로커 데이터 디렉터리 설정 `Oralce JDK` 사용시 데이터 디렉터리 변경 필요\ 컨플루언트 플랫폼 스크립트 -> `사용자 : cp-kafka` ```bash # 브로커 디렉터리 설정 $ sudo vi /etc/kafka/server.properties log.dirs=/var/lib/kafka/data # 로 수정, 기본 : /var/lib/kafka # 데이터 디렉터리 추가 $ sudo mkdir /var/lib/kafka/data # 컨플루언트 플랫폼 사용시 유저정보 권한 $ sudo chown cp-kafka:confluent /var/lib/kafka/data ``` * `여러대 동작하기 -> 추가 설정` `주키퍼, 카프카클러스터` -> 여러 대의 클러스로 동작시키기 위해서는 추가설정 필요 ```bash # 주키퍼 추가 설정 -> 아래 내용 추가 $ vi /etc/kafka/zookeeper.properties # 주키퍼 클러스터 초기 접속 및 동기 타임아웃 # tickTime이라는 파라미터를 단위로 계산 -> 기본값이 3000ms(3초) initLimit=10 # 30초(10 * 3초) syncLimit=5 # 15초(5 * 3초) # 클러스터를 구성하는 서버군의 정보 server.1=kafka-broker01:2888:3888 server.2=kafka-broker02:2888:3888 server.3=kafka-broker03:2888:3888 ``` > 서버 정보 형식 : server.=<서버 호스트명>:<서버 통신용 포트1>:<서버 통신용 포트2> myid : `주키퍼 클러스터 내`, 각 서버에 고유하게 부여된 `서버 번호` 호스트 kafka-broker01 -> 1, kafka-broker02 -> 2, kafka-broker03 -> 3 번에 해당 각 myid -> `/var/lib/zookeeper/myid`에 작성해야 함 **myid는 서버마다 다르기 떄문에 실행 명령도 서버마다 다름** * 각 서버별 명령어 실행 ```bash # kafka-broker01(myid=1) $ echo 1 | sudo -u cp-kafka tee -a /var/lib/zookeeper/myid # kafka-broker02(myid=2) $ echo 2 | sudo -u cp-kafka tee -a /var/lib/zookeeper/myid # kafka-broker03(myid=3) $ echo 3 | sudo -u cp-kafka tee -a /var/lib/zookeeper/myid ``` * 브로커 추가 설정 ```bash $ vi /etc/kafka/server.properties broker.id=<서버별로 정한 Broker ID> # 이미 작성된 내용 수정 broker.id.generation.enable=false # 추가 zookeeper.connect=kafka-broker01:2181,kafka-broker02:2181,kafka-broker03:2181 # 작성된 내용 수정 ``` > `log.dirs`를 변경한 상태(/var/lib/kafka -> /var/lib/kafka/data), 여기에서 그 파일에 추가로 여러 대의 서버에서 실행하기 위한 설정을 해야함 > > * broker.id : 브로커 ID 설정하기 위한 항목 > * broker.id 설정하지 않고, broker.id.generation.enable=true 설정시 -> 브로커 ID 자동 부여 > * `broker.id.generation.enable=false` 설정 -> broker.id 수동으로 설정 > * 브로커도 주키퍼의 myid와 마찬가지로 `브로커마다 고유 id`부여 필요 > * 브로커ID는 `정수값`으로 설정 해야함 > * 주키퍼의 `myid`와 동일할 필요는 없지만 여기에서 아래와 같이 설정 > * kafka-broker01 -> 1 > * kafka-broker02 -> 2 > * kafka-broker03 -> 3 ### 카프카 실행 및 동작 #### 카프카 클러스터 실행 실행 순서 : `주키퍼 실행 -> 브로커 실행` * 주키퍼 실행 카프카 클러스터 구축하고 있는 경우, 주키퍼가 설치된 모든 머신 실행. 순서는 상관없음 ```bash # 주키퍼 실행명령 $ sudo systemctl start confluent-zookeeper ``` * 브로커 실행 카프카 클러스터 구축하고 있는 경우, 브로커가 설치된 모든 머신 실행. 순서 상관없음 ```bash # 브로커 실행 $ sudo systemctl start confluent-kafka ``` * 출력 로그 * 주키퍼 : /var/log/kafka/zookeeper.out * 브로커 : /var/log/kafka/server.log #### 카프카 클러스터 동작 확인 카프카 빌트인 도구로 메시지를 전송하고, 제대로 송수신하는지 여부 확인이 가능함 * Kafka Console Producer * Kafka Console Consumer **동작확인** * 메시지 송수신을 위한 `토픽 작성` 필요 * `카프카 클라이언트`에서 명령 실행 * `first-test`라는 토픽 작성 * 주키퍼의 접속 정보가 다를 수 있기 때문에 구성한 서버에 따라 명령 다름 **서버 3대로 클러스터 구축한 경우** * \--zookeeper : 주키퍼의 접속정보 지정 * <호스트명>:<접속포트번호> -> 예) kafka-broker01:2181 .. * `1대의 경우 -> --zookeeper 호스트명:2181` * `/etc/kafka/server.properties` 의 `zookeeper.connect 필드값 확인` * \--create * 토픽작성 * 토픽목록 확인 : --list, 삭제 : --delete * \--topic * 작성하는 토픽이름 지정 * 예제는 first-test * 토픽이름 : 밑줄(\_), 마침표(.) 사용금지 * \--partitions * 토픽의 파티션수 * 예제는 3 * \--replication-factor * 토픽의 레플리카 복제본 수 * `카프카 클러스터의 브로커 수 이하여야한다` * 3대 -> 3이하, 1대 -> 1이하 (--replication-factor 3 or 1) * input : ```bash # 여러대 설정 $ kafka-topics --zookeeper kafka-broker01:2181,kafka-broker02:2181,kafka-broker02:2181 --create --topic first-test --partitions 3 --replication-factor 3 ``` * output : ``` Created topic "first-test" ``` **토픽이 정상 생성 확인** **주키퍼 접속 정보 확인 필수** ```bash # 토픽 생성 확인 $ kafka-topics --zookeeper kafka-broker01:2181,kafka-broker02:2181,kafka-broker02:2181 --describe --topic first-test ``` `--create -> --descirbe`으로 지정하여 명령. `--descirbe`은 토픽에 대한 상세정보 표시 옵션 Topic, PartitionCount, ReplicationFactor 표시된다. * output : ``` Topic:first-test PartitionCount:3 ReplicationFactor:1 Configs: Topic: first-test Partition: 0 Leader: 1 Replicas: 1 Isr: 1 Topic: first-test Partition: 1 Leader: 1 Replicas: 1 Isr: 1 Topic: first-test Partition: 2 Leader: 1 Replicas: 1 Isr: 1 ``` * Leader : 각 파티션의 현재 Leader Replicark 어떤 브로커에 존재하고 있는지 표시. `번호->브로커 ID` * Replicas : 각 파티션의 레플리카 보유하고 있는 `브로커 목록` * Isr : In-Sync-Replicas의 약자. 복제본 중 `Leader Replicas와 올바르게 동기가 실행된 복제본을 보유하고 있는 브로커 목록 표시`. 동기화가 실행되지 않은 레플리카는 Isr에 포함되지 않음. Leader Replicas 자신은 Isr에 포함됨 **메시지 송수신 확인** 카프카 빌트인 도구인 `Kafka Console Producer` 사용 하여 `콘솔에 데이터 입력->메시지 카프카 전송`\ 카프카 빌트인 도구인 `Kafka Console Consumer` 사용 하여 `토픽에서 메시지 취득` * 프로듀서 서버 ```bash # 전송 명령어 실행 $ kafka-console-pruducer --broker-list kafka-broker01:9092, kafka-broker02:9092, kafka-broker03:9092 --topic first-test ``` * \--broker-list * 메시지를 보내는 카프카 클러스터의 브로커 호스트명,포트번호 지정 * 호스트명:포트번호 형태로 지정 * 여러대 -> ,(쉼표)로 연결 * 카프카가 통신에 사용하는 `기본포트 9092 사용` * \--topic * 메시지 송신처 -> `토픽 지정` * 컨슈머 서버 ```bash # 메시지 취득 명령어 실행 $ kafka-console-consumer --bootstrap-server kafka-broker01:9092,kafka-broker02:9092,kafka-broker03:9092 --topic first-test ``` * \--bootstrap-server * 메시지 수신하는 `카프카 클러스터`의 `브로커 호스트명,포트번호 지정` * \--topic * 메시지를 `수신하는 토픽` **카프카 클러스터 종료** 실행순서 역순, `주키퍼 종료 -> 브로커 종료` ```bash # kafak (브로커) 종료 $ sudo systemctl stop confluent-kafka # 주키퍼 종료 $ sudo systemctl stop confluent-zookeeper ```