이번엔 지금까지 받은 데이터를 MongoDB에 저장하도록 한다.Docker에 MongoDB 설치프로젝트폴더> docker run -d --name mongodb -p 27017:27017 mongo현재 Docker에서는 mongodb와 kafka관련 항목이 실행 중이어야한다.이제 이전에 작성했던 kafka_consumer_server.py 파일을 수정한다. kafka_consumer_server.py 일부 수정...app = FastAPI() # FastAPI 인스턴스 생성mongo_client = MongClient("mongodb://localhost:27017/") # MongoDB 클라이언트 생성mongo_db = mongo_client["sensor_database"] # MongoDB 데이..

데이터를 전송할 에뮬레이터가 만들어졌기 때문에이제 Kafka에 저장된 데이터를 가져와서 처리하는 Consumer를 생성한다.Consumer의 역할은1. kafka topic로 부터 메시지를 읽어온다.2. Offset를 관리한다. 메시지를 읽어온 위치를 기억하는 역할을 담당한다. Offset 방식은 자동커밋과 수동커밋방식으로 구분된다.3. Consumer Group의 하나로 다중 인스턴스로 병렬처리가 가능하다.4. 부하 분산 및 재조정이 가능하다. Consumer Group에 속해 있기 때문에 하나의 Consumer이 죽어도 다른 Consumer에 재 분배되어 동작한다.5. 메시지 처리 후족 작업을 수행한다. DB저장, 알림 전송 등 비지니스 로직을 수행한다. 피자가게 예시로 들어보면..

앞서서는 환경설정이 끝났고이제 센서에서 초당 3200개의 데이터를 발생하는 에뮬레이터를 생성한다.우선 이번 페이지의 폴더구조는 다음과 같다.kafka-test├── docker-compose.yml ├── emulator └── sensor_emulator.py sensor_emulator.pyfrom kafka import KafkaProducerimport jsonimport timeimport randomimport threadingfrom datetime import datetime# kafka producer 설정producer = KafkaProducer( bootstrap_servers="localhost:9092", # Kafka 브로커 주소 value_serial..

회사에서 테스트 중에 있는 프로젝트 중센서 하나에서 초당 3200개의 데이터를 보내는걸 DB 저장하고 그래프로 시각화 해야 하는 내용이 있었다.단순히 thread를 돌려서 웹으로 처리하는 방식으로 친구들이 처리 하고 있는데.그건 1개의 센서에 대해서는 가능해 보이지만 다수의 센서에서 초당 3200개는 무리로 보였다.그래서 나도 테스트 작업 한번 해 보기로 하고 Python으로 작성을 시작하였다. 환경 구성 : Python, Kafka Python Client(kafka-python), FastAPI Docker 환경 : Kafka, zookeeper개발 : VSCode 나는 VSCode에서 터미널을 사용해서 진행한다.탐색기 등에서 프로젝트 폴더를 생성하고 VSCode의 파일에서 Open Folder.....