Posted inPython

MongoDB + Python: PyMongoによる基本的なCRUDから「究極」のAggregationまで

Python tutorial - IT technology blog
Python tutorial - IT technology blog

なぜMongoDBとPythonは「最強のコンビ」なのか？

アプリケーションのロジックを変更するたびにデータベースのマイグレーションを実行することに疲れているなら、MongoDBが救世主となります。柔軟なドキュメント構造により、厳密なスキーマ定義なしでJSON形式のようなデータを保存できます。これを、ミニマルな構文で知られるPythonと組み合わせることで、データ処理はかつてないほどスムーズになります。

多くのプロジェクトを経験してきた中で、PyMongoは最も信頼できるドライバーだと感じています。安定しているだけでなく、MongoDBの最新機能もフルサポートしています。この記事では、基本的な操作から、数百万件のレコードに対するクエリ最適化まで、実践的なテクニックを直接解説します。

クイックな環境構築

まず、MongoDBのインスタンス（DockerまたはMongoDB Atlas）が用意されていることを確認してください。次のコマンドでPyMongoライブラリをインストールします。

pip install "pymongo[srv]"

ヒント: "pymongo[srv]" を使用すると、DNSエラーを心配することなくCloud（Atlas）上のクラスターに接続するために必要な依存関係が自動的にインストールされます。

接続の実装とデータ処理

1. 標準的なコネクションプールの設定

無闇に接続を開かないようにしましょう。代わりに、コネクションプールを再利用するために接続処理を関数化し、サーバーリソースを節約します。

from pymongo import MongoClient
from pymongo.errors import ConnectionFailure

def get_database(uri="mongodb://localhost:27017/"):
    try:
        # 接続できない場合は5秒後にタイムアウト
        client = MongoClient(uri, serverSelectionTimeoutMS=5000)
        client.admin.command('ping')
        return client['dev_database']
    except ConnectionFailure:
        print("サーバーが応答しません。設定を確認してください！")
        return None

db = get_database()

2. CRUD：パフォーマンスをボトルネックにさせない

PyMongoの操作は、Pythonの dict を扱うのと非常によく似ています。しかし、パフォーマンスの差は関数の呼び出し方に現れます。

Insert（データの挿入）:

collection = db['users']

# 初心者のやり方：ループ内で1レコードずつ挿入
# プロのやり方：insert_manyを使用
users = [
    {"name": "Hoàng", "role": "DevOps", "skills": ["Python", "Docker"]},
    {"name": "An", "role": "Data", "skills": ["SQL", "Python"]}
]
result = collection.insert_many(users)
print(f"{len(result.inserted_ids)} 件のレコードを挿入しました。")

実際の数値: 10,000件のレコードを挿入するテストケースでは、insert_one を使うと約15〜20秒かかります。対照的に、insert_many は1秒もかかりません。常にバッチ処理を優先しましょう。

スマートな更新（Update）: ドキュメント全体を上書きするのではなく、$set や $push オペレーターを使用して必要なフィールドのみを更新します。

collection.update_one(
    {"name": "Hoàng"},
    {"$push": {"skills": "MongoDB"}}
)

3. Aggregation Pipeline：プロのようにデータを処理する

Aggregation（集計）はMongoDBの最も強力なツールです。データベース上で直接、計算、フィルタリング、データ変換を行うことができます。数GBものデータをPythonに取得してからPandasで処理する必要がなくなり、ネットワーク帯域の大幅な節約になります。

例：スキルごとの専門家数を集計する。

pipeline = [
    {"$unwind": "$skills"}, 
    {"$group": {
        "_id": "$skills",
        "total": {"$sum": 1}
    }},
    {"$sort": {"total": -1}}
]

for stat in collection.aggregate(pipeline):
    print(f"{stat['_id']}: {stat['total']} 人")

4. インデックス最適化の極意

インデックスの欠如だけでシステムがCPU過負荷に陥ることがあります。コレクションが100万件を超えると、インデックスのないクエリはデータベースにフルスキャンを強制することになります。

# 頻繁に検索するフィールドにインデックスを作成
collection.create_index([("role", 1)])

また、Projectionを活用しましょう。Emailだけが必要な場合、重いBio配列までMongoDBに返させる必要はありません。比較してみると、find({}, {"email": 1}) は通常の find({}) よりも30〜50%高速です。

監視とモニタリング（Monitoring）

クエリが最適に動作しているか確認するにはどうすればよいでしょうか？ explain() コマンドを使用しましょう。これはMongoDBがどのようにコマンドを実行したかの詳細を分解して示してくれます。

stats = collection.find({"role": "DevOps"}).explain()
print(stats['executionStats']['nReturned']) # 見つかったレコード数
print(stats['executionStats']['totalDocsExamined']) # スキャンされたレコード数

もし totalDocsExamined が nReturned よりも何倍も大きい場合は、すぐにインデックスを見直す必要があります。深夜にシステムが500エラーを吐くまで最適化を待たないようにしましょう。強力なバックエンドシステムの構築を応援しています！

Share:
Tags: