MongoDB Benchmark

from pymongo import MongoClient
import time

# MongoDB 연결
DATABASE_NAME = "benchmark_db"  # 데이터베이스 이름을 변수로 처리
client = MongoClient("mongodb://localhost:27017/")
db = client[DATABASE_NAME]
collection = db["benchmark_collection"]

# 테스트할 문서 개수 설정
NUM_DOCS = 10000  # 문서 수

def generate_document(doc_id):
    """문서 생성 함수"""
    return {"_id": doc_id, "value": doc_id, "index_value": doc_id}

def drop_database():
    """테스트 시작 전에 데이터베이스 드롭"""
    client.drop_database(DATABASE_NAME)  # 변수로 설정한 데이터베이스 이름으로 삭제
    print(f"[DATABASE] '{DATABASE_NAME}' 데이터베이스가 삭제되었습니다.")

def benchmark_insert():
    """MongoDB Insert 성능 테스트"""
    collection.delete_many({})  # 기존 데이터 삭제
    
    total_time = 0
    for i in range(NUM_DOCS):
        start = time.time()
        doc = generate_document(i)
        collection.insert_one(doc)
        end = time.time()
        total_time += (end - start)
    
    avg_time = total_time / NUM_DOCS
    print(f"[INSERT] {NUM_DOCS}개 문서 삽입 평균 시간: {avg_time:.4f}초")

def benchmark_find_without_index():
    """MongoDB Collection Scan (인덱스 없이) 성능 테스트"""
    total_time = 0
    count = 0
    for i in range(NUM_DOCS):
        start = time.time()
        doc = collection.find_one({"value": i})  # value로 조회
        end = time.time()
        total_time += (end - start)
        if doc:
            count += 1
    
    avg_time = total_time / NUM_DOCS
    print(f"[FIND] 컬렉션 스캔 (인덱스 없이) 평균 조회 시간: {avg_time:.4f}초, 조회된 문서 수: {count}")

def benchmark_create_index():
    """MongoDB 인덱스 생성"""
    start = time.time()
    collection.create_index([("index_value", 1)])  # 'index_value' 필드에 인덱스 생성
    end = time.time()
    
    print(f"[INDEX] 'index_value' 필드에 인덱스 생성 시간: {end - start:.4f}초")

def benchmark_find_with_index():
    """MongoDB Index Scan (인덱스 사용) 성능 테스트"""
    benchmark_create_index()  # 인덱스 생성 후 테스트
    
    total_time = 0
    count = 0
    for i in range(NUM_DOCS):
        start = time.time()
        doc = collection.find_one({"index_value": i})  # index_value로 조회
        end = time.time()
        total_time += (end - start)
        if doc:
            count += 1
    
    avg_time = total_time / NUM_DOCS
    print(f"[FIND] 인덱스 스캔 (인덱스 사용) 평균 조회 시간: {avg_time:.4f}초, 조회된 문서 수: {count}")

def benchmark_update_without_index():
    """MongoDB Update (인덱스 없이) 성능 테스트"""
    total_time = 0
    for i in range(NUM_DOCS):
        start = time.time()
        collection.update_one({"_id": i}, {"$set": {"value": 999999}})  # value 업데이트
        end = time.time()
        total_time += (end - start)
    
    avg_time = total_time / NUM_DOCS
    print(f"[UPDATE] 컬렉션 스캔 (인덱스 없이) 평균 업데이트 시간: {avg_time:.4f}초")

def benchmark_update_with_index():
    """MongoDB Update (인덱스 사용) 성능 테스트"""
    total_time = 0
    for i in range(NUM_DOCS):
        start = time.time()
        collection.update_one({"index_value": i}, {"$set": {"index_value": 888888}})  # index_value 업데이트
        end = time.time()
        total_time += (end - start)
    
    avg_time = total_time / NUM_DOCS
    print(f"[UPDATE] 인덱스 스캔 (인덱스 사용) 평균 업데이트 시간: {avg_time:.4f}초")

def benchmark_delete_without_index():
    """MongoDB Delete (인덱스 없이) 성능 테스트"""
    total_time = 0
    for i in range(NUM_DOCS):
        start = time.time()
        collection.delete_one({"_id": i})  # _id로 삭제
        end = time.time()
        total_time += (end - start)
    
    avg_time = total_time / NUM_DOCS
    print(f"[DELETE] 컬렉션 스캔 (인덱스 없이) 평균 삭제 시간: {avg_time:.4f}초")

def benchmark_delete_with_index():
    """MongoDB Delete (인덱스 사용) 성능 테스트"""
    total_time = 0
    for i in range(NUM_DOCS):
        start = time.time()
        collection.delete_one({"index_value": i})  # index_value로 삭제
        end = time.time()
        total_time += (end - start)
    
    avg_time = total_time / NUM_DOCS
    print(f"[DELETE] 인덱스 스캔 (인덱스 사용) 평균 삭제 시간: {avg_time:.4f}초")

if __name__ == "__main__":
    drop_database()  # 테스트 시작 전에 데이터베이스 드롭
    benchmark_insert()  # 데이터 삽입
    benchmark_find_without_index()  # 인덱스 없이 조회
    benchmark_find_with_index()  # 인덱스 사용하여 조회
    benchmark_update_without_index()  # 인덱스 없이 업데이트
    benchmark_update_with_index()  # 인덱스를 사용한 업데이트
    benchmark_delete_without_index()  # 인덱스 없이 삭제
    benchmark_delete_with_index()  # 인덱스를 사용한 삭제

모든 collection compact

mongo your_database_name -u your_user -p your_pass --authenticationDatabase admin --quiet --eval 'db.getCollectionNames().forEach(function(c){ print("Compacting: " + c); printjson(db.runCommand({compact: c})); })'