📌 토픽 생성

[jms0522]

logstash conf 에서 kafka topic을 설정

컨테이너 내부 kafka-topic.sh 있는 곳에서 실행하거나 docker exec -it를 사용해서 외부에서 실행

토픽 목록 확인

 kafka-topics.sh --list --bootstrap-server localhost:9092 

토픽 생성

kafka-topics.sh --create \
  --topic <토픽이름> \
  --partitions <파티션수> \
  --replication-factor <복제본수> \
  --bootstrap-server localhost:9092

정보 확인

kafka-topics.sh --describe \
  --topic <토픽이름> \
  --bootstrap-server localhost:9092

데이터 들어오는지 확인

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic new_test_topic --from-beginning

Python 코드로 토픽 생성 및 목록 확인 (kakfa-python)

설치

 pip install kafka-python

 from kafka.admin import KafkaAdminClient, NewTopic
 from kafka import KafkaConsumer

  # Kafka 클라이언트 설정
  admin_client = KafkaAdminClient(
      bootstrap_servers="localhost:9092",
      client_id='kafka-python-example'
  )

# 토픽 생성 함수
  def create_topic(topic_name, num_partitions=3, replication_factor=1):
      topic = NewTopic(name=topic_name, num_partitions=num_partitions, replication_factor=replication_factor)
      admin_client.create_topics(new_topics=[topic], validate_only=False)
      print(f"Topic '{topic_name}' created successfully.")

# 토픽 목록 확인 함수
  def list_topics():
      consumer = KafkaConsumer(bootstrap_servers="localhost:9092")
      topics = consumer.topics()
      print("Topics:", topics)

# 토픽 생성
  create_topic("logstash-topic")

# 토픽 목록 확인
    list_topics()

docker-compose를 통해 시작시 토픽 생성도 가능하다.

environment:
  KAFKA_CREATE_TOPICS: "logstash-topic:3:1"  # 토픽 생성: logstash-topic, 파티션 3, 복제본 1

이런식으로 시작시 생성할 수 있음.

스크립트를 통해서 생성도 가능하다.

command: >
  sh -c "
  /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties &
  sleep 5;
  /opt/kafka/bin/kafka-topics.sh --create --topic logstash-topic --partitions 3 --replication-factor 1 --if-not-exists --bootstrap-server localhost:9092;
  /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092;
  wait
  "

이런식으로 정해두면 스크립트로 실행된다. 단 sh파일은 미리 존재해야함.

[jms0522]

설정 파일 위치

/opt/kafka_2.13-2.8.1/config/server.properties