MongoDB 변경 스트림 및 Python을 사용한 실시간 데이터 처리-파이썬 튜토리얼-php.cn

Real-Time Data Processing with MongoDB Change Streams and Python

소개

MongoDB의 변경 스트림을 사용하면 애플리케이션이 실시간 데이터 변경에 즉시 반응할 수 있습니다. 이 블로그 게시물에서는 이론에 너무 깊이 들어가지 않고도 Python을 사용하여 변경 스트림을 설정하고 사용하는 방법을 보여 드리겠습니다. 먼저 삽입에 초점을 맞춘 다음 이를 다른 이벤트 유형으로 확장하여 데이터베이스 이벤트를 수신하는 간단한 프로그램을 만들어 보겠습니다.

변경 스트림 시작하기

변경 스트림을 사용하면 앱이 삽입이나 업데이트와 같은 특정 데이터베이스 이벤트를 수신하고 즉시 응답할 수 있습니다. 사용자가 자신의 프로필을 업데이트하는 시나리오를 상상해 보세요. 변경 스트림을 사용하면 사용자가 페이지를 새로 고칠 필요 없이 앱 전체에 이 변경 사항을 즉시 반영할 수 있습니다. 이 기능 이전에는 지속적으로 데이터베이스를 폴링하거나 MongoDB Oplog 테일링과 같은 복잡한 방법을 사용해야 했습니다. 변경 스트림은 보다 사용자 친화적인 API를 제공하여 이를 단순화합니다.

변경 스트림 없이는 어떻게 되나요?

인보이스를 업로드하는 API가 있다고 가정해 보겠습니다. 흐름은 고객이 MongoDB에 송장 이미지를 업로드한 다음 AI로 정보를 추출하고 송장을 업데이트하는 것입니다. 송장 업로드 코드의 예는 다음과 같습니다.

from pymongo import MongoClient

class MongoDatabase:
    def __init__(self, config_path: str):
        # Load the YAML configuration file using the provided utility function
        self.config_path = config_path
        self.config = read_config(path=self.config_path)

        # Initialize MongoDB connection
        self.client = MongoClient(self.config['mongodb']['uri'])
        self.db = self.client[self.config['mongodb']['database']]
        self.collection = self.db[self.config['mongodb']['collection']]

    def create_document(self, data: Dict[str, Any]) -> str:
        # Insert a new document and return the automatically generated document ID
        result = self.collection.insert_one(data)
        return str(result.inserted_id)
    def update_document_by_id(self, document_id: str, data: Dict[str, Any]):
        try:
            self.collection.update_one({"_id": document_id}, {"$set": data})
        except PyMongoError as e:
            print(f"Error updating document: {e}")

로그인 후 복사

먼저 만일을 대비해 pymongo를 클래스 안에 포장하겠습니다 :))

@app.post("/api/v1/invoices/upload")
async def upload_invoice(request: Request):
    try:
        # Parse JSON body
        body = await request.json()
        img = body.get("img")
        user_uuid = body.get("user_uuid")

        if not img or not is_base64(img):
            return JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"status": "error", "message": "Base64 image is required"},
            )

        # Generate invoice UUID
        current_time = datetime.now(timezone.utc)
        img = valid_base64_image(img)       
        invoice_document = {
            "invoice_type": None,
            "created_at": current_time,
            "created_by": user_uuid,
            "last_modified_at": None,
            "last_modified_by": None,
            "status": "not extracted",
            "invoice_image_base64": img,
            "invoice_info": {}
        }

        invoice_uuid = mongo_db.create_document(invoice_document)
        print('Result saved to MongoDB:', invoice_uuid)
        mongo_db.update_document_by_id(invoice_uuid, {"invoice_uuid": invoice_uuid})

        return JSONResponse(
            status_code=status.HTTP_201_CREATED,
            content={"invoice_uuid": invoice_uuid, "message": "Upload successful"}
        )
    except Exception as e:
        # Handle errors
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"status": "error", "message": str(e)}
        )

로그인 후 복사

합리적인 질문은 다음과 같습니다. 업데이트하기 전에 AI 모델이 이미지를 처리할 때까지 기다리지 않으시겠습니까? 문제는 처리하는 데 약 4~5분 정도 소요된다는 점이며, 사용자 경험에 영향을 미치고 싶지 않습니다.

카프카는 어때요?

또 다른 옵션은 Kafka를 사용하는 것입니다. 이미지를 Kafka 주제에 게시하면 다른 서비스가 데이터를 처리할 수 있습니다.

장점:

업로드 및 처리 서비스를 분리합니다.
대규모 실시간 데이터 처리에 효율적입니다.
향상된 사용자 경험: 사용자는 인보이스를 업로드한 후 즉시 응답을 받습니다. 처리는 비동기적으로 처리됩니다.

단점:

더 복잡해졌습니다.
Kafka 인프라 설정 및 유지 관리가 필요합니다.
소규모 애플리케이션에는 과잉일 수 있습니다.

다음은 Kafka를 사용하여 송장 업로드 프로세스를 처리하는 방법을 보여주는 기본 구현입니다.

사용자가 API 엔드포인트를 통해 송장을 업로드합니다. 송장 이미지는 MongoDB에 저장되며 추가 처리를 위해 메시지가 Kafka 주제로 전송됩니다.

from kafka import KafkaProducer
import json
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
from datetime import datetime, timezone

app = FastAPI()

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

@app.post("/api/v1/invoices/upload")
async def upload_invoice(request: Request):
    try:
        body = await request.json()
        img = body.get("img")
        user_uuid = body.get("user_uuid")

        if not img or not is_base64(img):
            return JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"status": "error", "message": "Base64 image is required"},
            )

        current_time = datetime.now(timezone.utc)
        img = valid_base64_image(img)       
        invoice_document = {
            "invoice_type": None,
            "created_at": current_time,
            "created_by": user_uuid,
            "status": "not extracted",
            "invoice_image_base64": img,
        }

        # Save the document to MongoDB
        invoice_uuid = mongo_db.create_document(invoice_document)
        mongo_db.update_document_by_id(invoice_uuid, {"invoice_uuid": invoice_uuid})

        # Send a message to Kafka topic
        producer.send('invoice_topic', invoice_document)
        producer.flush()

        return JSONResponse(
            status_code=status.HTTP_201_CREATED,
            content={"message": "Invoice upload received and will be processed"}
        )
    except Exception as e:
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content={"status": "error", "message": str(e)}
        )

로그인 후 복사

Kafka 소비자는voice_topic을 듣습니다. 메시지를 받으면 송장을 처리하고(예: 이미지에서 정보 추출) MongoDB에서 해당 문서를 업데이트합니다.

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'invoice_topic',
    bootstrap_servers=['localhost:9092'],
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

for message in consumer:
    invoice_document = message.value

    # Process the invoice, extract information, and update the document in MongoDB
    invoice_uuid = invoice_document["_id"]
    extracted_data = extract_invoice_data(invoice_document["invoice_image_base64"])

    mongo_db.update_document_by_id(invoice_uuid, {
        "invoice_info": extracted_data, 
        "status": "extracted"
    })

    print(f"Processed and updated invoice: {invoice_uuid}")

로그인 후 복사

흐름 요약:

인보이스 업로드: 사용자가 API를 통해 인보이스를 업로드합니다.
MongoDB에 저장: 송장 문서가 MongoDB에 저장됩니다.
Kafka로 메시지 보내기: 송장 세부정보가 포함된 메시지가 Kafka 주제(invoice_topic)로 전송됩니다.
Kafka 소비자가 송장을 처리합니다. Kafka 소비자는 Invoice_topic을 듣고 송장을 처리하며 추출된 정보로 MongoDB의 해당 문서를 업데이트합니다.

와, 제가 이걸 직접 썼다니 믿을 수가 없네요! 실제로 관련된 노력을 강조합니다. MongoDB, Kafka, Invoice 서비스라는 세 가지 서비스를 관리하고 구성하는 복잡성도 고려하지 않은 것입니다.

MongoDB 변경 스트림을 사용한 송장 처리

다음은 변경 스트림에 의해 트리거되는 송장 처리를 처리하는 추가 메서드와 기능을 포함하여 MongoDB 변경 스트림을 보여주기 위해 Markdown으로 다시 작성된 전체 코드입니다.

문서 생성 및 변경 스트림 수신과 같은 데이터베이스 작업을 처리하는 MongoDB 래퍼 클래스를 만드는 것부터 시작하겠습니다.

from pymongo import MongoClient
from pymongo.errors import PyMongoError
from typing import Dict, Any
import threading
import yaml

class MongoDatabase:
    # Same code as before #

    def process_invoice(self, invoice_document: Dict[str, Any]):
        """Process the invoice by extracting data and updating the document in MongoDB."""
        try:
            # Simulate extracting information from the invoice image
            extracted_data = extract_invoice_data(invoice_document["invoice_image_base64"])
            invoice_uuid = invoice_document["_id"]

            # Update the invoice document with the extracted data
            self.update_document_by_id(invoice_uuid, {"invoice_info": extracted_data, "status": "extracted"})
            print(f"Processed and updated invoice: {invoice_uuid}")
        except Exception as e:
            print(f"Error processing invoice: {str(e)}")

    def start_change_stream_listener(self):
        """Start listening to the change stream for the collection."""
        def listen():
            try:
                with self.collection.watch() as stream:
                    for change in stream:
                        if change['operationType'] == 'insert':
                            invoice_document = change['fullDocument']
                            print(f"New invoice detected: {invoice_document['_id']}")
                            self.process_invoice(invoice_document)
            except PyMongoError as e:
                print(f"Change stream error: {str(e)}")

        # Start the change stream listener in a separate thread
        listener_thread = threading.Thread(target=listen, daemon=True)
        listener_thread.start()

로그인 후 복사

쉽게 하기 위해 MongoDatabase 클래스 내에 process_invoice를 추가합니다. 하지만 다른 곳에 맡겨야 합니다

업로드 API는 원본과 같아야 합니다.

mongo_db = MongoDatabase(config_path='path_to_your_config.yaml')
mongo_db.start_change_stream_listener()

@app.post("/api/v1/invoices/upload")
async def upload_invoice(request: Request):
    try:
        # Parse JSON body
        body = await request.json() 
        # same code as before

로그인 후 복사

흐름 요약:

사용자 업로드 송장: 사용자가 API를 통해 송장을 업로드합니다.
MongoDB에 저장: 송장 문서가 MongoDB에 저장됩니다.
MongoDB 변경 스트림 트리거됨: MongoDB 변경 스트림이 새 문서 삽입을 감지합니다.
송장 처리: 변경 스트림은 송장을 처리하고 추출된 정보로 MongoDB의 문서를 업데이트하는 process_invoice 함수를 트리거합니다.