Data Engineering

Docker를 활용한 카프카(Kafka) Cluster 환경 구축

kindof

·

2022. 9. 26. 11:58

이번 포스팅에서는 Docker 환경에서 Producer, Broker, Consumer가 동작하는 Kafka 환경을 구축해보겠습니다. 클러스터 구축은 단일 브로커로 동작하는 환경과 여러 개의 Kafka 클러스터가 동작하는 환경을 각각 살펴보겠습니다.

 

먼저 본격적인 실습에 앞서 Kafka 환경을 이해하는 데 필요한 몇 가지 개념을 짚어보겠습니다.

 

1. Kafka Concepts

Kafka는 이벤트 스트리밍을 분산 처리하는 툴로써 이벤트(메시지)를 생산하는 Producer와 이를 처리하는 Consumer사이에서 동작합니다.

 

이전에는 Producer - Consumer 사이에서 메시지가 직접적으로 왔다갔다 하는 환경이었다면, 이제는 Kafka가 그 사이의 결합도를 낮춰줌으로써 복잡한 애플리케이션 환경을 보다 단순하게 매개해주고 대용량 메시지 처리와 안정성을 확보할 수 있게 된 것이죠.

 

이러한 Kafka의 장점에 대해서는 앞으로의 포스팅을 통해 직접 실습해보면서 이야기해보도록 하겠습니다. 우선, Kafka 공식 문서를 참고해서 몇 가지 용어에 대한 정의부터 살펴보겠습니다.

 

1-1. Event

Event는 애플리케이션의 비즈니스 로직에서 발생하는 어떤 일련의 사건을 의미합니다. 쉽게 말해 Kafka에서 처리하는 메시지들인데요. Event는 아래와 같이 Key, Value, timestamp, metadata 등으로 이루어져 있는 객체입니다.

Event key: "Alice"
Event value: "Made a payment of $200 to Bob"
Event timestamp: "Jun. 25, 2020 at 2:06 p.m."

- https://kafka.apache.org/documentation/#intro_concepts_and_terms

 

참고로 Kafka Producer는 Broker들에게 데이터를 보낼 때 Byte Array 형태로 처리하기 때문에 아래와 같이 Producer는 Serializer를 사용하여 메시지를 변환해야 하고, Consumer는 Deserializer를 사용하여 다시 원하는 자료형으로 데이터를 변환해주어야 합니다.

Producer - Serializer
Consumer - Deserializer

 

1-2. Producer, Consumer

Producer와 Consumer는 그 이름에서 의미를 대충 짐작할 수 있습니다. Producer는 이벤트(메시지)를 발행하여 Broker에게 전달하고, Consumer는 이러한 이벤트를 구독하여 처리하는 녀석입니다.

 

과거에는 Producer가 직접적으로 Consumer에게 메시지를 제공했다면 현재는 Kafka가 Producer에게 받은 이벤트를 저장하고 Consumer가 저장된 이벤트를 가져가서 처리하는 구조가 된 것이죠.

 

덕분에 Producer는 Consumer의 환경에 신경쓰지 않고 이벤트를 생산하며, Consumer 역시 뒤에서 설명할 Partition, Replication 등의 Kafka에서 제공하는 특징들로 인해 메시지의 고가용성과 안정성을 확보할 수 있게 되었습니다.

 

1-3. Topics, Partition, Broker

Topic은 Producer로부터 전달받은 이벤트를 분류해서 저장하는 공간이라고 이해하면 됩니다.

 

하나의 Topic에는 여러 Partition이 존재할 수 있으며 각 Partition은 Broker란 녀석이 관리하고 있습니다. 그리고 여러 Partition에 다수의 Producer가 이벤트를 저장할 수 있고, 다수의 Consumer가 동시에 이벤트에 접근할 수도 있습니다.

 

이것이 바로 이벤트 스트리밍 환경에서 Kafka는 분산 플랫폼으로 역할한다는 의미가 됩니다. 또한 이렇게 저장되고 사용되는 이벤트들은 삭제되지 않고 일정 기간동안 보존되도록 Config를 작성해줄 수도 있죠. 

https://d2.naver.com/helloworld/0853669

 

한편, Topic > Partition 에 의한 이벤트 저장 장소의 다원화, 분리를 통해 클라이언트는 Read/Write를 동시에 수행할 수도 있는 장점도 가지게 됩니다.

 

또한 Topic에 저장된 이벤트들은 다른 곳에 복사(Replicated)됨으로써 데이터의 유실을 방지할 수도 있게 되는데, 일반적으로 여러 Topic과 Broker가 존재할 때 Replication factor는 '3'으로 설정함으로써 데이터를 보관하고 있습니다.

https://kafka.apache.org/documentation/#intro_concepts_and_terms

 

1-4. Zookeeper

Zookeeper는 분산 애플리케이션을 위한 코디네이션 서비스입니다. 우리가 사용하는 Kafka 노드들의 관리와 동기화 등을 Zookeeper라는 녀석이 대행해준다는 뜻인데요.

 

Kafka는 Zookeeper를 사용해서 Broker, Topic, Partition 등에 대한 메타데이터를 저장하고 이용합니다.

 

조금 더 구체적으로는 Zookeeper 안에는 znode가 존재하고 znode에 Key-Value 형태로 메타데이터가 저장되고 이를 통해 Kafka는 개별적인 클라이언트가 되어 Zookeeper 서버들과 상태 정보 등을 주고 받는 것입니다.

https://zookeeper.apache.org/doc/current/zookeeperOver.html

 

여기까지 이해하셨다면 Kafka가 어떤 툴인지, 어떻게 동작하는지 대충 감은 잡으셨으리라 생각합니다. 이제 Docker 환경에서 아주 간단한 Kafka 환경 구축을 해보는 실습을 해보겠습니다.

 

참고로 실습 환경은 Linux Mint 21, x64-64입니다. Mac OS와 과정은 거의 비슷하지만 CLI가 조금씩 다르니 주의해주세요.

 

2. 단일 브로커 Kafka 환경

기본적으로 Docker 설치는 완료되었다고 가정하겠습니다. 아래와 같이 설치된 Docker 버전을 확인할 수 있습니다.

$ docker -v
>> Docker version 20.10.18, build b40c2f6

 

다음으로는 아래와 같이 docker-compose.yml 파일을 다운로드 받겠습니다. 

$ curl --silent --output docker-compose.yml https://raw.githubusercontent.com/confluentinc/cp-all-in-one/7.2.1-post/cp-all-in-one/docker-compose.yml

$ cat docker-compose.yml 
---
version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.2.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-server:7.2.1
    hostname: broker
    container_name: broker

...
...

  rest-proxy:
    image: confluentinc/cp-kafka-rest:7.2.1
    depends_on:
      - broker
      - schema-registry
    ports:
      - 8082:8082
    hostname: rest-proxy
    container_name: rest-proxy
    environment:
      KAFKA_REST_HOST_NAME: rest-proxy
      KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092'
      KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
      KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'

docker-compose.yml은 위 내용처럼 여러 컨테이너 이미지(zookeeper, broker 등)들을 하나의 yml 파일로 관리하여 실행할 수 있도록 합니다. 

 

이제 다운로드 받은 docker-compose 파일을 실행하고, 컨테이너의 동작을 확인해보겠습니다.

$ docker compose up -d
$ docker ps

 

docker-compose 파일 실행을 완료하면 아래와 같은 내용을 볼 수 있고, 실제로 실행된 컨테이너 목록들을 확인할 수 있습니다.

docker-compose.yml 실행
docker ps

 

이제 실행된 컨테이너들을 기반으로 Kafka 환경을 구축하겠습니다. 먼저 Kafka를 다운로드 받고 심볼릭 링크를 지정해줌으로써 버전 관리를 할 수 있도록 설정하겠습니다.

$ wget https://dlcdn.apache.org/kafka/3.2.3/kafka_2.12-3.2.3.tgz
$ tar zxf kafka_2.12-3.2.3.tgz
$ ln -s kafka_2.12-3.2.3.tgz kafka
$ ls -l

 

그리고 kafka가 설치된 경로에서 아래 명령어를 통해 브로커 한 개짜리(1 Partition, 1 Replication) 토픽을 생성하겠습니다.

$ kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic peter-test01 --partitions 1 --replication-factor 1 --create

 

참고로 --bootstrap-server는 브로커의 주소를 의미하며, 그 뒤에 토픽 이름과 Partition, Replication의 수를 나열합니다.

 

콘솔 창에서 Producer, Consumer의 동작을 확인하기 위해 다른 두 개의 터미널에서 아래 명령어를 각각 실행합니다.

$ kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic peter-test01
$ kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic peter-test01

Producer - Consumer

그러면 콘솔 창에서 확인할 수 있는 결과와 같이 peter-test01 이라는 토픽을 통해 Producer의 이벤트가 쌓이고, Consumer는 해당 이벤트를 Read하고 있음을 알 수 있습니다.

 

이 상황을 아주 간단한 그림으로 정리해보면 아래와 같습니다.

 

 

3. 다중 Broker Kafka 환경 구축

여러 Broker가 존재하는 Kafka 환경 구축도 단일 Broker 환경과 크게 다르지 않습니다. 마찬가지로 docker compose를 이용할건데요.

 

아래 링크를 통해 3개의 Zookeeper, 3개의 Kafka Broker 환경을 구축할 수 있는 yml 파일을 작성합니다(단일 Kafka 환경에서 사용하던 컨테이너들은 docker compose stop CLI를 통해 종료해주세요).

version: '2'
services:
  zookeeper-1:
    image: confluentinc/cp-zookeeper:7.2.1
    environment:
      ZOOKEEPER_SERVER_ID: 1
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
    ports:
      - "22181:2181"

  zookeeper-2:
    image: confluentinc/cp-zookeeper:7.2.1
    environment:
      ZOOKEEPER_SERVER_ID: 2
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
    ports:
      - "32181:2181"

  zookeeper-3:
    image: confluentinc/cp-zookeeper:7.2.1
    environment:
      ZOOKEEPER_SERVER_ID: 3
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_INIT_LIMIT: 5
      ZOOKEEPER_SYNC_LIMIT: 2
    ports:
      - "42181:2181"
  
  kafka-1:
    image: confluentinc/cp-kafka:7.2.1
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    ports:
      - 29092:29092
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      
  kafka-2:
    image: confluentinc/cp-kafka:7.2.1
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    ports:
      - "39092:39092"
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:9092,PLAINTEXT_HOST://localhost:39092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT

  kafka-3:
    image: confluentinc/cp-kafka:7.2.1
    depends_on:
      - zookeeper-1
      - zookeeper-2
      - zookeeper-3
    ports:
      - "49092:49092"
    environment:
      KAFKA_BROKER_ID: 3
      KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-3:9092,PLAINTEXT_HOST://localhost:49092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT

몇 가지 설정값에 대한 설명을 짚어보겠습니다.

 

[Zookeeper 설정값]

  • ZOOKEEPER_CLIENT_PORT : Zookeeper 접속을 위한 포트(Default = 2181)입니다. 해당 포트를 통해 Kafka 클라이언트들이 연결됩니다.
  • ZOOKEEPER_TICK_TIME : Zookeeper가 사용하는 시간에 대한 기본 측정 단위(2초)입니다.
  • ZOOKEEPER_INIT_LIMIT : Leader, Follower가 연결 시도를 하는 최대 횟수입니다.
  • ZOOKEEPER_SYNC_LIMIT : Follower가 리더와 동기화되기 위한 제한값입니다.

[Kafka 설정값]

  • depends_on :  해당 컨테이너(Zookeeper)들이 정상적으로 생성되어야한다는 제약(의존성)입니다.
  • KAFKA_ZOOKEEPER_CONNECT : Broker들의 Metadata를 Zookeeper에 저장하기 위한 위치를 명시합니다. <HOST>:<PORT> 형식으로 작성해줍니다. 
  • KAFKA_ADVERTISED_LISTENERS : Producer, Consumer에게 노출할 주소를 말합니다. 설정하지 않을 경우 server.properties에 명시된 listeners 설정을 따르게 됩니다. Docker에서 실습을 진행 중이기 때문에 localhost 작성으로 충분합니다. 다만, 다른 외부의 IP에서 접근을 해야 한다면 해당 부분에 내용을 추가해주시면 됩니다.
  • KAFKA_INTER_BROKER_LISTENER_NAME: Broker 간 통신에 사용할 Listener를 정의합니다. 

 

이제 kafka가 설치된 경로에서 server.properties에서 listeners 부분의 주석을 해제합니다. Broker가 사용할 호스트와 포트 번호를 지정한 것입니다.

server.properties

이제 작성한 yml 파일을 실행하고 컨테이너를 확인해보겠습니다. -d 옵션을 주게 되면 zookeeper, kafka를 daemon 환경에서 실행합니다.

$ docker compose -f docker-compose-multi.yml up -d
$ docker ps

 

이제 Payment라는 토픽을 하나 생성하고 하나의 토픽은 2개의 Partition, 3개의 Replication을 가지도록 하겠습니다.

$ docker compose exec kafka-1 kafka-topics --create --topic payment --bootstrap-server kafka-1:9092 --replication-factor 3 --partitions 2
$ docker compose exec kafka-1 kafka-topics --describe --topic payment --bootstrap-server kafka-1:9092

Topic Creation

이제 kafka-1, kafka-2 클러스터에 Consumer를 생성하고, kafka-1에 연결되는 Producer를 생성해주겠습니다. Replication factor 값이 3으로 설정되었기 때문에 Producer에서 전달된 이벤트는 kafka-1,2,3에 모두 복제됩니다.

// kafka-1를 바라보는 Consumer 생성
$ docker compose exec -it kafka-1 bash
$ kafka-console-consumer --topic payment --bootstrap-server kafka-1:9092

// kafka-2를 바라보는 Consumer 생성
$ docker compose exec -it kafka-2 bash
$ kafka-console-consumer --topic payment --bootstrap-server kafka-2:9092

// Producer 생성
$ docker compose exec -it kafka-1 bash
$ kafka-console-producer --topic payment --broker-list kafka-1:9092

 

그리고 터미널을 통해 Producer에서 메시지를 보내보면 예상한 것처럼 두 곳의 Consumer에서 메시지를 읽어오는 것을 확인할 수 있습니다!

왼쪽 두 개(Consumer) / 오른쪽(Producer)

 

4. 나가면서

이번 포스팅에서는 Kafka의 기본적인 개념에 대해 정리해보고, Docker 위에서 간단한 Kafka 환경을 구축해봤습니다. 앞으로의 포스팅에서는 실제 코드를 작성해가면서 Kafka에 대해 공부해보려고 합니다.

 

감사합니다.

 

5. Reference

https://d2.naver.com/helloworld/0853669

https://devocean.sk.com/blog/techBoardDetail.do?page=&query=&ID=163709&searchData=Kafka&subIndex=&idList=%5B164096%2C+164016%2C+164007%2C+163980%2C+163970%2C+163709%2C+158649%5D 

https://kafka.apache.org/documentation/#intro_concepts_and_terms

https://devocean.sk.com/blog/techBoardDetail.do?ID=164016 

 

'Data Engineering' 카테고리의 다른 글

Kafka Producer Message Batching과 몇 가지 성능 튜닝 옵션들  (0) 2023.07.01
Kafka Consumer, ConsumerGroup 그리고 Rebalancing  (0) 2023.06.11
자주 사용하는 Kafka CLI 정리하기  (0) 2023.02.24
Kafka offset reset을 통한 이벤트 Consuming  (0) 2023.02.01
파이썬 스크립트로 Kafka Partition Reassign 자동화하기(Kafka URP 대응기)  (1) 2022.12.14

Data Engineering 카테고리와 연관된 콘텐츠