Kafka Producer 및 Consumer 실습해보기 with Python (+ksqlDB) - TIL230713

📚 KDT WEEK 15 DAY 4 TIL

• Kafka 기본 프로그래밍
• Topic 생성 및 파라미터 설정
• Consumer 생성 및 파라미터 설정
• ksqlDB

---

🟥 Kafka 기본 프로그래밍

 

나의 Kafka 설치 환경은 conduktor사가 제공하는 Repo를 git clone해서 가져왔다.

이 글은 이 환경을 기반으로 설명한다.

참고 : https://sunhokimdev.tistory.com/66

Kafka 알아보고 설치해보기 - TIL230712

 

터미널에서 간단하게 Kafka를 사용해보는 실습을 해보자.

 

1. Topics 리스트하기

 

docker ps # Broker에 해당하는 컨테이너의 ID 확인
docker exec -it Broker_ContainerID sh # 쉘로 실행

 

docker-compose.yml에 작성된 Broker의 이름을 찾아 그 컨테이너 ID를 사용하면 된다.

나와 같은 환경으로 Kafka를 구축했다면, kafka1의 컨테이너 ID를 사용하면 된다.

 

Broker 컨테이너 쉘에 접속했다.

 

쉘에서 다음의 명령어를 입력하면 현재 가지고있는 Topic을 리스트할 수 있다.

kafka-topics --bootstrap-server kafka1:9092 --list

 

kafka-topics으로 Topic에 관련된 명령들을 수행할 수 있다.

--bootstrap-server는 사용할 Broker를 지정하는 명령어인데, 내 환경에서는 Broker가 도커 컨테이너에 들어가있다.

따라서 Broker 컨테이너의 호스트 이름과 Broker가 돌고있는 포트번호를 넣어 사용한다.

 

어떤 topic이 있는지 확인할 수 있다.

 

docker-connect-*는 Kafka Connect가 사용하는 일종의 DB같은 Topic들이다.

__consumer_offsets은 어떤 consumer가 어떤 topic의 어떤 데이터까지 읽었는지 기록되는 topic이다.

 

여기서 리스트된 이름을 확인해서, 지우고 싶은 토픽은 다음 명령어로 지울 수 있다.

 

kafka-topics --bootstrap-server kafka1:9092 --delete --topic 토픽이름

 

2. Producer와 Consumer 사용해보기

 

쉘에서 각각 Producer와 Consumer를 사용하는 방법은 다음과 같이 입력하면 된다.

 

kafka-console-producer --bootstrap-server kafka1:9092 --topic test_console # Producer
kafka-console-consumer --bootstrap-server kafka1:9092 --topic test_console # Consumer

 

test_console은 사용하고자 하는 topic의 이름이 된다.

터미널 두 개를 동시에 열어서 각각 쉘에 진입한 후, Producer와 Consumer를 위 명령어로 하나씩 사용해보자.

그리고 Producer에서 값을 입력할 때마다 실시간으로 Consumer에 반영되는 것을 확인할 수 있다!

 

Producer

 

Consumer

 

웹 UI에서도 Topic에 들어간 데이터가 보인다.

 

---

 

🟦 Topic 생성 및 Topic 파라미터 설정

이제까지는 자동으로 Topic이 생성되도록하여 따로 파라미터를 설정하지 않았다.

Topic의 파라미터를 설정함으로써 파티션의 수와 복제본의 수도 조정할 수 있다.

 

Producer 파이썬 파일안에서 새로운 Topic을 생성하고 데이터를 넣는 실습을 해보았다.

 

먼저 producer로 사용할 파이썬 파일을 하나 만들었다.

 

fake_person_producer.py

import uuid
import json
from typing import List
from person import Person # 모델이 있는 person.py에서 Person 클래스 참조

from faker import Faker # 가짜 사람 데이터를 만드는데 사용
from kafka.admin import NewTopic
from kafka.errors import TopicAlreadyExistsError # Topic이 이미 존재할 경우 에러
from kafka import KafkaAdminClient
from kafka.producer import KafkaProducer

# 새로운 Topic을 생성하는 함수
# 인자는 순서대로 Broker 리스트, Topic 이름, 파티션 수, 복제본 수
def create_topic(bootstrap_servers, name, partitions, replica=1):
    client = KafkaAdminClient(bootstrap_servers=bootstrap_servers)
    try:
        topic = NewTopic(
            name=name,
            num_partitions=partitions,
            replication_factor=replica)
        client.create_topics([topic])
    except TopicAlreadyExistsError as e: # Topic이 이미 존재하는 경우 에러 발생하지만, 그냥 pass하도록 구현
        print(e)
        pass
    finally:
        client.close()


def main():
    topic_name = "fake_people"
    bootstrap_servers = ["localhost:9092"]

    # fake_people 이름으로 4개의 파티션을 갖는 새로운 Topic 생성
    create_topic(bootstrap_servers, topic_name, 4)

    faker = Faker()
    producer = KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        client_id="Fake_Person_Producer", # Producer에 이름 주기
        
    )

    # faker 라이브러리를 사용해서 랜덤한 사람 100명 데이터 Topic에 보내기
    for _ in range(100):
        person = Person(id=str(uuid.uuid4()), name=faker.name(), title=faker.job().title())

        producer.send(
            topic=topic_name,
            key=person.title.lower().replace(r's+', '-').encode('utf-8'), # 소문자로 바꾸고, 공백이 여러 개인 경우 '-'로 바꿈
            value=person.json().encode('utf-8')) # json으로 바꾸고 utf-8로 인코딩

    producer.flush()

if __name__ == '__main__':
    main()

 

faker로 생성한 title(직업)을 기준으로 Key로 만들었다.

나중에 Key가 동일한 데이터끼리 묶여 파티션으로 나누고, 이와 관련된 효율적인 작업이 가능할 수도 있을 것이다.

 

person.py

from pydantic import BaseModel

class Person(BaseModel):
    id: str
    name: str
    title: str

 

이 두개의 파일을 Kafka가 설치된 폴더에 넣고 실행해보자.

 

python fake_persone_producer.py

 

실행 결과가 웹 UI에 반영된 모습

 

실행이 완료된 후, 웹 UI에서 fake_people Topic을 확인해보면 들어간 데이터를 확인할 수 있다.

Key, Value 모두 설정한 값대로 들어갔다.

 

---

  

🟩 Consumer 생성 및 파라미터 설정

 

Consumer의 수보다 Partition의 수가 더 많은 경우, Partition은 라운드 로빈 방식으로 Consumer들에게 할당된다.

이때, 기본적으로 한 Partition은 한 Consumer에게만 할당된다.

 

먼저, 메시지를 읽는 보장 방식에는 Message Processing Guarantee에는 세 가지가 있다.

 

• Exactly Once : 각 Message가 Consumer에게 정확히 한 번만 전달된다는 것을 보장한다. > 구현 어려움
• At Least Once : Consumer에게 적어도 한 번 이상 전달되도록 보장하지만, 중복될 수도 있다.
• At Most Once(Default) : 메시지가 전달될 수도 있고, 아닐 수도 있지만 중복 전달되는 경우는 없다. 

 

Message Processing Guarantee

 

이 방식들 중에서 나의 환경이나 데이터들을 고려하여 한 가지를 선택해서 구현하게 된다.

 

Consumer는 enable_auto_commit 옵션이 True인 경우와 False 경우로 나누어서 만들었다.

 

> enable_auto_commit = True인 Consumer의 경우

autocommit_consumer.py

import json
from kafka.consumer import KafkaConsumer


def key_deserializer(key):
    return key.decode('utf-8')


# Producer의 Serializer와 반대로 동작
def value_deserializer(value):
    return json.loads(value.decode('utf-8'))


def main():
    topic_name = "fake_people"
    bootstrap_servers = ["localhost:9092"]
    consumer_group_id = "fake_people_group"

    consumer = KafkaConsumer(
        bootstrap_servers=bootstrap_servers,
        group_id=consumer_group_id,
        key_deserializer=key_deserializer,
        value_deserializer=value_deserializer,
        auto_offset_reset='earliest', # 처음 실행될 때 가장 앞에 있던 기록부터 읽어옴
        enable_auto_commit=True
    )

    consumer.subscribe([topic_name])
    for record in consumer:
        print(f"""
            Consumed person {record.value} with key '{record.key}'
            from partition {record.partition} at offset {record.offset}
        """)


if __name__ == '__main__':
    main()

 

subscribe한 Topic들로부터 Producer의 직렬화한 데이터를 받아 역직렬화하고, print한다.

auto_commit 옵션이 True이기 때문에, offset과 Partition 정보들을 알아서 Commit한다.

 

알아서 offset 정보를 업데이트하기 때문에 매번 offset이 다르게 나온다.

 

> enable_auto_commit = False인 Consumer의 경우

manualcommit_consumer.py

import json

from kafka import TopicPartition, OffsetAndMetadata # 두 개의 새로운 모듈 필요
from kafka.consumer import KafkaConsumer


def key_deserializer(key):
    return key.decode('utf-8')


def value_deserializer(value):
    return json.loads(value.decode('utf-8'))


def main():
    topic_name = "fake_people"
    bootstrap_servers = ["localhost:9092"]
    consumer_group_id = "manual_fake_people_group" # 아까와는 다른 그룹

    consumer = KafkaConsumer(
        bootstrap_servers=bootstrap_servers,
        group_id=consumer_group_id,
        key_deserializer=key_deserializer,
        value_deserializer=value_deserializer,
        auto_offset_reset='earliest',
        enable_auto_commit=False # autocommit이 False
    )

    consumer.subscribe([topic_name])
    for record in consumer:
        print(f"""
            Consumed person {record.value} with key '{record.key}'
            from partition {record.partition} at offset {record.offset}
        """)

        topic_partition = TopicPartition(record.topic, record.partition) # Topic의 파티션 정보를 읽기
        offset = OffsetAndMetadata(record.offset + 1, record.timestamp) # offset 위치 +1하고, timestamp도 업데이트
        consumer.commit({
            topic_partition: offset # 파티션 위치와 offset 정보 COMMIT
        })

if __name__ == '__main__':
    main()

 

새로운 두 가지 모듈을 추가하고, 몇 줄정도 추가되었다.

개발자가 직접 Commit이 가능하도록 코드를 작성해야 한다.

 

TopicPartition 모듈로 partition 정보를 담고, OffsetAndMetadata 모듈로 offset 정보를 담아 각각 Key와 Value로 Commit 했다.

 

직접 Commit을 넣어주었더니, 아까와 동일한 결과가 나타났다.

 

동일한 결과가 나타난다면 성공이다.

 

웹 UI에서 생성된 Consumer Group을 확인할 수 있다.

 

Consumer Groups

 

autocommit_consumer.py로 생성된 fake_people_group은 아까 동작을 멈췄기 때문에 State가 EMPTY로 나타난다.

manualcommit_consumer.py로 생성된 manual_fake_people_group은 지금도 동작을 하고 있어서,

State가 STABLE이고, Members도 하나 있는 것으로 나타나고 있다.

 

 

manual_fake_people_group의 partiton 정보

 

manual_fake_people_group의 Partition 정보에도 어떤 결과가 있었는지 대략 나타난다.

 

---

 

🟨 ksqlDB

REST API나 ksql 클라이언트 툴을 사용하면 Topic을 테이블처럼 SQL 조작이 가능해진다.

CREATE STREAM 명령어로 새로운 Stream을 하나 생성해서, 테이블처럼 다룰 수 있다.

이 방법 또한 하나의 Consumer로 동작하는 것으로 취급해서, Consumer Group과 Consumer ID를 자동으로 갖게 된다.

하지만 Kafka Topic에 별개의 ksqlDB를 위한 Topic이 존재하여 그 밑에 저장되므로 Consumer Group 쪽에 나타나지는 않는다.

 

docker ps # ksqldb-server 컨테이너 ID 찾기
docker exec -it ContainerID sh # ksqldb-server 쉘에 접속
ksql # ksql 접속

 

ksql에 접속된 모습

 

여기서 아까 생성했던 Topic의 데이터를 확인해보자.

 

CREATE STREAM my_stream (id STRING, name STRING, title STRING) with (kafka_topic='fake_people', value_format='JSON');
SELECT * FROM my_stream;

 

fake_people Topic 정보를 토대로 id, name, title 데이터를 가져온다.

 

SELECT 쿼리문의 결과

 

여기서 Producer가 데이터를 생성할 때의 시간 데이터를 새로운 컬럼으로 붙여보자.

 

SELECT *, ROWTIME FROM my_stream;

 

ROWTIME을 사용하면 그 역할을 해준다.

 

참고로 뒤에 EMIT CHANGES 을 추가하면 아까 읽어온 데이터 뒤로 새로운 데이터만 읽어온다.

 

LIMIT도 똑같이 동작한다.