1. Kafka - Topic

Kafka에서 가장 기본이 되는 Topic은 메시지(=레코드, 이벤트, 데이터)들이 저장되는 논리적인 개념이다.

 

각 Topic은 용도에 맞게 만들어서 데이터를 저장하게 된다.

ex) 서버 모니터링, 어플리케이션 로그 데이터 등..

 

Kafka 사전 지식

 Kafka에는 Topic과 연관된 여러 개념들이 있다. 나중에 설명하겠지만 미리 설명해보자면,

1. producer: 메시지를 생산 및 Kafka의 Topic으로 보냄

2. consumer: Topic의 메시지를 소비

3. consumer group: 협력하는 consumer들의 모임(분산, 병렬처리를 가능하게 함)

 

kafka arhcitecture

4. commit log: 추가만 가능하고 변경 불가능한 Data Structure. 메시지는 항상 로그 끝에 추가된다.(한번 추가되면 변경이 불가능)

5. Offset: commit log의 메시지 위치(LOG-END-OFFSET, CURRENT-OFFSET), 하나의 partition에서만 의미를 가진다.

 

commit log

 

Topic의 기본 개념

Topic을 한마디로 설명하자면 목적별로 데이터를 나누어 놓은 카테고리이다.

서점으로 예를 들면 스릴러, 로맨스 등등 각각의 분류에 맞게 책을 정리 해놓으면(Produce), 독자들은 그 카테고리에 가서 책을 찾으면 된다(Consume).

(위의 서점의 예시가 원하는 데이터를 찾는다는 점에서 100% 부합하는 설명은 아니지만 이해를 돕기 위해서...)

 

다시 Kafka로 돌아와 설명을 하자면, A라는 Cluster의 서버에서 나오는 Log 데이터는 A-Cluster-Topic, B라는 Cluster 서버에서 나오는 Metric은 B-Cluster-Metric 이런식으로 Topic을 분리하게 된다면, Consumer가 각 목적에 맞는 Topic을 Consume 하면 데이터의 분리를 용이하게 할 수 있는 것이다. 

 

Partition

Partition은 Topic에서 가장 중요한 개념이라고 생각 된다. 이유는 Kafka를 이용하는 이유는 대규모의 데이터를 병렬처리 한다는 것에서 속도적인 이점을 볼 수 있다는 것인데, Partition은 그것을 가능하게 하는 개념이기 때문이다.

 

Partition은 기본적으로 하나의 Commit Log라고 할 수 있고, Kafka의 병렬처리(Throughput 향상)의 기본이 되는 것은 다수의 Partition 사용하여 한다고 할수 있다.

 

또한 각각의 Partition은 서로 독립적인 관계이다.

 

Segment

메시지가 저장되는 실제 물리 파일이고 실제로 byte code로 저장되어 있다.

---

• Topic 생성시 Partiton 개수를 지정한다(개수 변경이 가능하나 운영시에는 권장하지 않는다. 이유에 대해선 나중에 설명하기로 하자)
• Partition들은 Broker에 분산되어 Segment 파일로 구성되어 있다.
• Rolling Strategy: log.segment.byte(def 1G), log.roll.hours(def 168h)로 파일의 크기를 정한다.  Segment가 새로 생성되는 조건이다.
• Partition에저장된 데이터는 변경 불가능(immutable)하다.

---

Topic관련 CLI

# server dns my-kafka1,2,3


# topic 생성 명령 
$ kafka-topics.sh --create \
--bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \
--partitions 3 \
--replication-factor 1 \
--config retention.ms=172800000 \
--topic hello.kafka
# --config 옵션은 명령에 포함하지 않은 추가적인 옵션이 실행가능하다.

# topic 리스트 명령어 
$ kafka-topics.sh --list \
--bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \

# topic 상세 조회
$ kafka-topics.sh --describe \
--bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \
--topic hello.topic

# topic제거
$ kafka-topics.sh --delete
--bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \
--topic hello.topic

# record 삭제
# record refer from json file
$ cat > delete-topic.json <EOF
> {"partitions": [{"topic": "test", "partiton":0, "offset":50}], "verion":1}
> EOF

$ ./kafka-delete-records.sh \
--bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \
--offset-json-file delete-topic.json

Topic의 Config 변경

Topic의 config 변경의 경우에는 bin/kafka-config.sh를 이용해서 바꾸는 것이 좋다.

--alter 옵션을 줄 수도 있자만 추후에는 삭제될 예정이라고 한다.

# topic의 retention.ms 설정 변경 방법
$ kafka-config.sh --bootstrap-server my-kafka1:9092,my-kafka2:9092,my-kafka3:9092 \
--entity-type topics \
--entity-name hello.kafka \
--alter --add-config retention.ms=8640000