[Week15 스트리밍 데이터 처리 : Kafka] TIL 63일차 Docker를 이용한 Kafka 환경 구축(Feat. Google Cloud)

이번 시간에는 yaml파일을 사용하여 Docker에서 Kafka를 실행해 보자.

Ⅰ. GCP에 Docker컨테이너로 Kafka 환경 구축하기

1. GCP VM 인스턴스 설정

이전의 airflow와 비슷한 과정이다.

full-stack Kafka yaml파일을 실행할 것이기 때문에 컴퓨팅 자원이 많은 vm엔진을 선택하였다.

• c2-standard-4
• Ubuntu-20.04 30GB

 

2. 서버에 Docker 설치

① 시스템의 패키지 목록을 최신 상태로 업데이트

② Docker를 다운로드하고 설치하기 위해 필요한 추가적인 패키지를 설치(apt-transport-https, ca-certificates, curl, software-properties-common)

③ Docker의 공식 GPG 키를 다운로드하여 APT 패키지 관리자에 추가

④ Docker 패키지를 제공하는 저장소를 시스템에 추가

⑤ 새로 추가된 Docker 저장소 정보를 포함하여 패키지 목록을 업데이트

⑥ 설치 가능한 Docker 버전의 목록을 표시 ⑦ 앞서 설정한 저장소를 통해 Docker Community Edition (CE)를 설치 + Docker 엔진 활성화

sudo apt update
sudo apt install apt-transport-https ca-certificates curl software-properties-common
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -
sudo add-apt-repository "deb [arch=amd64] https://download.docker.com/linux/ubuntu bionic stable"
sudo apt update
apt-cache policy docker-ce
sudo apt install docker-ce

 

3. 서버에 Docker-compose 설치

① Docker compose 패키지 다운로드 (2 버전 이상을 다운로드하자. 1 버전은 오류가 많이 난다)

② 다운로드한 Docker Compose 바이너리 파일에 실행 권한을 부여

③ 현재 사용자를 Docker 그룹에 추가

sudo curl -L https://github.com/docker/compose/releases/download/v2.4.1/docker-compose-`uname -s`-`uname -m` -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose
sudo usermod -aG docker $USER

 

4. Docker image로 환경 구축하기

1) 도커 이미지가 있는 깃허브 레포지토리 클론

git clone https://github.com/conduktor/kafka-stack-docker-compose

 

2) Kafka 도커 이미지 다운로드 및 컨테이너 실행

kafka-stack-docker-compose 폴더로 이동한 뒤, 도커 이미지를 다운로드하고 컨테이너를 실행시킨다.

cd kafka-stack-docker-compose
docker-compose -f full-stack.yml pull
docker compose -f full-stack.yml up

 

 

Ⅱ. Kafka 웹페이지 접속하기

gcp-vm머신 외부 ip:8080 주소로 접속하면 다음과 같이 kafka conduktor를 웹으로 접속할 수 있다.

로그인 아이디, 패스워드는 다음과 같다.

login: admin@admin.io password: admin

 

Ⅲ. python을 이용해 Kafka 작업수행하기

1. python 모듈 설치하기

pip3 install kafka-python

 

2. Producer 만들기 예제

로컬 Kafka 인스턴스를 연결하는 KafkaProducer 객체를 생성

전송하려는 데이터를 json 문자열로 변환한 다음 UTF-8로 인코딩하여 직렬화하는 방법을 정의

from json import dumps
from kafka import KafkaProducer

producer = KafkaProducer(
   bootstrap_servers=[Broker의 목록],
   value_serializer=lambda x: dumps(x).encode('utf-8')
)
data={key:value}
producer.send('topic 이름', value=data)

 

Topic이 잘 만들어진 것을 웹 UI에서 확인할 수 있다.

Producer가 생성한 Topic

3. Consumer 만들기

consumer = KafkaConsumer(
 '소비할 topic이름',
 bootstrap_servers=[brokerip:8080의 집합],
 auto_offset_reset='earliest',
 enable_auto_commit=True,
 group_id='동일한 자원을 소비하는 consumer을 그룹으로 지정',
 value_deserializer=lambda x: loads(x.decode('utf-8'))
)

- auto_offset_reset

• earlist : 큐에 저장된 순서대로 데이터를 읽겠다
• latest : 큐에 새로 생긴 데이터부터 읽겠다 

- enable_auto_commit

• True : kafka consumer object가 알아서 offset값을 기록
• False : kafka consumer object가 알아서 offset값을 기록하지 않음(보통 현업에서는 위험해서 잘 사용하지 않음. 대신, Kafka consumer 함수중 commit을 사용해서 명시적으로 위치를 기록)