[Queue] Kafka 적용기

1.Kafka란?

메시지 큐 서비스
-고성능 분산 이벤트 스트리밍 플랫폼

 

2.카프카, 쥬키퍼 쿠버네티스환경 적용

kubectl apply -f zookeeper.yaml
kubectl apply -f kafka.yaml

 

#적용확인
kubectl get pods

 

 

3.카프카 토픽생성

 

 

kubectl get pods --all-namespaces

#위명령어 통해 {kafka-pod} 확인
#ex: kafka-deployment-7985656cd5-vhz7d

 

#Pod 내부의 컨테이너 셸에 접속
kubectl exec -it pod/{kafka-pod} -- /bin/bash
# ex: kubectl exec -it kafka-deployment-7985656cd5-vhz7d -- /bin/bash

#토픽생성
kafka-topics --create --bootstrap-server localhost:29092 --replication-factor=1 --partitions=1 --topic=category-match-in
kafka-topics --create --bootstrap-server localhost:29092 --replication-factor=1 --partitions=1 --topic=category-match-out


#기존카프카토픽 확인하려면
kafka-topics --list --bootstrap-server localhost:29092

 

Q. 왜 두개의 토픽을 생성하는걸까? 

 

두 개의 토픽을 사용하는 이유는 데이터의 흐름(Flow)을 관리하기 위해입니다.
category-match-in과 category-match-out은 일반적으로 입력 데이터와 출력 데이터를 처리하기 위한 토픽으로 사용됩니다.

1. category-match-in (입력 토픽)

• 메시지를 만들어 Kafka 클러스터에 보낼 때 사용

2. category-match-out (출력 토픽)

• Kafka 클러스터에서 메시지를 소비하고 처리 결과를 저장할 때 사용

 

4.카프카 produce , consoume 테스트

 

kubectl apply -f pod.yaml
kubectl exec -it pod/test-pod -- /bin/bash

vi main.py
python main.py

 

5.kafka worker

docker build . -t 0hyeon/lecture_worker:0.1.0
#image확인
docker images

docker push 0hyeon/lecture_worker:0.1.0

FROM python:3.9

RUN apt-get update && apt-get install -y vim && apt-get install -y default-jdk
ENV JAVA_HOME /usr/lib/jvm/default-java

RUN pip install confluent-kafka==2.3.0 pandas==2.1.3 konlpy==0.6.0 tensorflow==2.13.0

 

 

6.쥬키퍼없는 최신 카프카

services:
  kafka-controller:
    image: apache/kafka:3.8.1
    container_name: kafka-controller
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-controller:9093,2@kafka-controller-2:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    networks:
      - codewithyu

  kafka-controller-2:
    image: apache/kafka:3.8.1
    container_name: kafka-controller-2
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: controller
      KAFKA_LISTENERS: CONTROLLER://:9093
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-controller:9093,2@kafka-controller-2:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    networks:
      - codewithyu

  kafka_broker:
    image: apache/kafka:3.8.1
    container_name: kafka_broker
    ports:
      - "29092:9092"
    environment:
      KAFKA_NODE_ID: 3
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: "PLAINTEXT://:19092,PLAINTEXT_HOST://:9092"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka_broker:19092,PLAINTEXT_HOST://localhost:29092"
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-controller:9093,2@kafka-controller-2:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - kafka-controller
      - kafka-controller-2
    networks:
      - codewithyu

  kafka_broker_1:
    image: apache/kafka:3.8.1
    container_name: kafka_broker_1
    ports:
      - "39092:9092"
    environment:
      KAFKA_NODE_ID: 4
      KAFKA_PROCESS_ROLES: broker
      KAFKA_LISTENERS: "PLAINTEXT://:19092,PLAINTEXT_HOST://:9092"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka_broker_1:19092,PLAINTEXT_HOST://localhost:39092"
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-controller:9093,2@kafka-controller-2:9093
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
    depends_on:
      - kafka-controller
      - kafka-controller-2
    networks:
      - codewithyu

networks:
  codewithyu:

 

 

7.Producer 생성 (데이터생성)