[AWS/Kafka] EC2에 kafka 설치하기

kafka description

💡 각 DB와 시스템 사이에 kafka를 도입하여 보내는 쪽(producer)과 받는 쪽 (consumer) 신경 쓸 필요 없이 kafka에서 중개역할 수행

• controller
  • 각 broker에게 담당 파티션 할당 수행
  • broker 동작 모니터링 관리
• kafka broker (=server)
  • 실행된 Kafka application server를 의미함
  • producerdprp message를 받아서 offset(consumer에서 메시지를 어디부터 어디까지 읽었는지 저장하는 값) 할당
    • 만약 offset=4 인 경우, offset 0,1,2,3은 메시지를 읽은 것으로 추정
    • partition내의 한 칸, offset, index
• apache zookeeper (=message 관리자)
  • 여러 대의 broker를 중재하고 연결하는 역할
  • 데이터 연계 시 broker에게 데이터를 전달하는 코디네이터 역할을 수행함
  • broker의 메타 데이터 저장 및 공유

• kafka와 데이터를 주고받기 위해 사용되는 Java library실제 데이터 통신이 필요할 때, 원하는 db와 시스템을 end to end 방식으로 연계하는 것이 아닌 kafka cluster를 두고 kafka application 간에 통신을 하는것

---

EC2 instance

• ubuntu - 22.04 LTS (HVM), SSD volume type | t2.medium | EBS 8Gib

$ mv jenny-pem.pem /Users/Desktop
$ ssh -i {your-pem.pem} ubuntu@{ip}

java install

$ sudo apt-get update 
$ sudo apt-get install openjdk-8-jdk

$ java -version

kafka install

$ wget <https://archive.apache.org/dist/kafka/2.5.0/kafka_2.12-2.5.0.tgz>
#wget 설치 

$ tar xvf kafka_2.12-2.5.0.tgz
#tar 아카이브 압축 해제

server.properties <kafka 설정 파일 수정>

advertised.listeners=PLAINTEXT://<ec2-ip>:9092
zookeeper.connect=localhost:2181

Zookeeper 실행

• 주키퍼는(Zookeeper)는 분산 코디네이션 서비스를 제공하는 오픈소스 프로젝트
• 클러스터에서 구성 서버들끼리 공유되는 데이터를 유지하거나 어떤 연산을 조율하기 위해 주로 사용

/home/ubuntu/kafka_2.12-2.5.0/bin/zookeeper-server-start.sh -daemon /home/ubuntu/kafka_2.12-2.5.0/config/zookeeper.properties

jps
# 주키퍼 프로세스 확인

kafka 실행

$ /home/ubuntu/kafka_2.12-2.5.0/bin/kafka-server-start.sh -daemon /home/ubuntu/kafka_2.12-2.5.0/config/server.properties

$ jps
# kafka 프로세스 확인

Topic 생성 [jenny라는 이름의 Topic]

./kafka-topics.sh --create --topic jenny --bootstrap-server localhost:9092

producer / consumer 실행

./kafka-console-producer.sh --broker-list localhost:9092 --topic jenny
#producer

# command + T 

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic jenny --from-beginning
#consumer

Kafka 커넥트/ 커넥터

• 커넥트 : 커넥터를 동작하도록 실행해주는 프로세스, 파이프라인으로 동작하는 커넥터를 동작하기 위해서는 반드시 커넥트를 실행시켜야 함
• 커넥터 : 실질적으로 데이터를 처리하는 코드가 담긴 jar 패키지, 특정 동작을 하는 코드 뭉치

💡 단일 실행모드 커넥트
간단한 데이터 파이프라인을 구성하거나 개발용으로 주로 사용

분산 모드 커넥트
실질적으로 상용에 활용하고 싶다면 분산 모드 커넥트로 구성해서 운영함
여러 개의 프로세스(커넥트)를 한개의 클러스터로 묶어서 운영하는 방식
일부 커넥트에 장애가 발생하더라도 파이프라인 복구 가능

💡 싱크 커넥터
어딘가로 데이터를 싱크한다 → 특정 저장소에 데이터를 저장하는 역할 = 컨슈머의 역할과 비슷

소스 커넥터
데이터베이스로부터 데이터를 가져와서 토픽에 넣는 역할, 즉 프로듀서의 역할을 하는 것과 비슷

Kafka Topic

• 파티션을 늘릴 순 있지만 줄일 수는 없음

---

Error

1. 카프카 자체의 property & 환경변수 에러 예상

 

2. kafka-import systax error

source venv-test-airflow/bin/activate
pip uninstall kafka
pip install kafka-python

 

3. kafka 미구동

##jps로 확인했을때 kafka 안 켜지면 
$ kill -9 112224

###topic 삭제
./kafka-topics.sh --delete --topic topic_iris --bootstrap-server localhost:9092

 

4. Permission denied (publickey)

chmod 400 jenny-pem.pem ssh -i "{your_pem.pem}" ubuntu@ec2-{your-ip}.ap-northeast-2.compute.amazonaws.com yes