Python脚本MongoDB查询如何建索引

wen 实用脚本 2026-07-19 2

本文目录导读：

使用pymongo创建索引
索引类型详解
索引管理
实际应用示例
索引优化建议
索引管理工具
注意事项

在Python中为MongoDB创建索引,主要有以下几种方式：

使用pymongo创建索引

基础索引创建

from pymongo import MongoClient
from pymongo import ASCENDING, DESCENDING
# 连接MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['mycollection']
# 创建单字段索引（升序）
collection.create_index([("field_name", ASCENDING)])
# 创建单字段索引（降序）
collection.create_index([("field_name", DESCENDING)])
# 创建复合索引
collection.create_index([
    ("field1", ASCENDING),
    ("field2", DESCENDING)
])
# 创建唯一索引
collection.create_index(
    [("email", ASCENDING)],
    unique=True
)
# 创建文本索引（用于全文搜索）
collection.create_index([
    ("content", "text"),
    ("title", "text")
])
# 创建TTL索引（自动过期删除）
collection.create_index(
    [("created_at", ASCENDING)],
    expireAfterSeconds=3600  # 1小时后自动删除
)

索引类型详解

常用索引类型

from pymongo import MongoClient
from bson.code import Code
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['mycollection']
# 单字段索引
collection.create_index("username")  # 简化写法，默认升序
# 复合索引（多字段组合）
collection.create_index([
    ("status", 1),  # 1表示升序
    ("created_at", -1)  # -1表示降序
])
# 多键索引（针对数组字段）
collection.create_index("tags")  # 无需特殊处理
# 哈希索引（用于分片）
collection.create_index(
    [("user_id", "hashed")]
)
# 地理空间索引
collection.create_index([("location", "2dsphere")])
# 稀疏索引（只索引包含该字段的文档）
collection.create_index(
    [("email", 1)],
    sparse=True
)
# 部分索引（只索引满足条件的文档）
collection.create_index(
    [("status", 1)],
    partialFilterExpression={
        "status": {"$gte": 1}
    }
)

索引管理

查看和管理索引

from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['mycollection']
# 获取所有索引信息
indexes = collection.indexes()
for index in indexes:
    print(f"Index name: {index.get('name')}")
    print(f"Index keys: {index.get('key')}")
    print(f"Index options: {index.get('options', {})}")
    print("---")
# 查看集合的索引
index_info = collection.index_information()
for name, info in index_info.items():
    print(f"Index: {name}")
    print(f"Details: {info}")
# 删除指定索引
collection.drop_index("index_name_here")
# 删除所有索引（除了_id默认索引）
collection.drop_indexes()
# 检查索引是否已存在
existing_indexes = collection.index_information()
if "my_index" not in existing_indexes:
    collection.create_index([("field", 1)])

实际应用示例

用户系统索引策略

from pymongo import MongoClient, ASCENDING, DESCENDING
from datetime import datetime
client = MongoClient('mongodb://localhost:27017/')
db = client['user_system']
users_collection = db['users']
# 用户集合的索引设计
def create_user_indexes():
    # 1. 登录查询：按邮箱查询用户
    users_collection.create_index(
        [("email", ASCENDING)],
        unique=True,  # 邮箱唯一
        background=True  # 后台创建，不阻塞
    )
    # 2. 按用户名查询
    users_collection.create_index(
        [("username", ASCENDING)],
        unique=True
    )
    # 3. 按角色和状态查询
    users_collection.create_index([
        ("role", ASCENDING),
        ("status", ASCENDING),
        ("created_at", DESCENDING)
    ])
    # 4. 地理位置索引（如果用户有位置信息）
    users_collection.create_index([("location", "2dsphere")])
    # 5. 全文搜索索引（用于搜索用户）
    users_collection.create_index([
        ("username", "text"),
        ("display_name", "text"),
        ("bio", "text")
    ])
    # 6. 过期时间索引（临时账户）
    users_collection.create_index(
        [("expires_at", ASCENDING)],
        expireAfterSeconds=0  # 文档过期
    )
# 创建索引
create_user_indexes()

日志系统索引策略

from pymongo import MongoClient, ASCENDING
from datetime import datetime, timedelta
client = MongoClient('mongodb://localhost:27017/')
db = client['logging_system']
logs_collection = db['logs']
def create_log_indexes():
    # 1. 按时间和级别查询日志
    logs_collection.create_index([
        ("timestamp", ASCENDING),
        ("level", ASCENDING)
    ])
    # 2. 按用户ID查询日志
    logs_collection.create_index([
        ("user_id", ASCENDING),
        ("timestamp", ASCENDING)
    ])
    # 3. TTL索引：自动删除30天前的日志
    logs_collection.create_index(
        [("timestamp", ASCENDING)],
        expireAfterSeconds=30 * 24 * 3600  # 30天
    )
    # 4. 部分索引：只索引错误和警告级别的日志
    logs_collection.create_index(
        [("timestamp", ASCENDING)],
        partialFilterExpression={
            "level": {"$in": ["ERROR", "WARNING"]}
        }
    )

索引优化建议

索引创建最佳实践

from pymongo import MongoClient
import time
client = MongoClient('mongodb://localhost:27017/')
db = client['ecommerce']
collection = db['products']
def create_optimized_indexes():
    # 1. 使用后台创建索引（避免阻塞）
    collection.create_index(
        [("sku", ASCENDING)],
        unique=True,
        background=True,  # 后台创建
        name="sku_unique_index"  # 命名索引便于管理
    )
    # 2. 创建覆盖查询的复合索引
    # 如果一个查询只需要price和name字段，可以创建覆盖索引
    collection.create_index([
        ("category", ASCENDING),
        ("price", ASCENDING)
    ])
    # 3. 监控索引性能
    start_time = time.time()
    # 执行查询
    results = list(collection.find(
        {"category": "electronics"},
        {"price": 1, "name": 1}
    ).sort("price", 1))
    query_time = time.time() - start_time
    # 4. 使用hint强制执行特定索引（测试用）
    results = list(collection.find(
        {"category": "electronics"}
    ).hint([("category", ASCENDING)]))
    # 5. 创建复合索引时的字段顺序建议
    # 等值条件 -> 排序条件 -> 范围条件
    collection.create_index([
        ("status", ASCENDING),  # 等值条件
        ("created_at", DESCENDING),  # 排序条件
        ("price", ASCENDING)  # 范围条件
    ])

索引管理工具

批量创建和管理索引

from pymongo import MongoClient
from pymongo.errors import OperationFailure
class IndexManager:
    def __init__(self, connection_string):
        self.client = MongoClient(connection_string)
    def create_indexes_batch(self, db_name, collection_name, indexes_config):
        """批量创建索引"""
        db = self.client[db_name]
        collection = db[collection_name]
        results = []
        for config in indexes_config:
            try:
                result = collection.create_index(
                    config['keys'],
                    **config.get('options', {})
                )
                results.append({
                    'name': config.get('name', 'unnamed'),
                    'status': 'created',
                    'index_name': result
                })
            except OperationFailure as e:
                results.append({
                    'name': config.get('name', 'unnamed'),
                    'status': 'failed',
                    'error': str(e)
                })
        return results
    def get_index_usage_stats(self, db_name, collection_name):
        """获取索引使用统计"""
        db = self.client[db_name]
        stats = db.command("collstats", collection_name)
        return stats.get("indexStats", [])
# 使用示例
manager = IndexManager('mongodb://localhost:27017/')
# 索引配置
indexes = [
    {
        'keys': [("email", 1)],
        'options': {'unique': True, 'name': 'email_unique'}
    },
    {
        'keys': [("username", 1)],
        'options': {'name': 'username_index'}
    },
    {
        'keys': [("status", 1), ("created_at", -1)],
        'options': {'name': 'status_created_complex'}
    }
]
# 批量创建索引
results = manager.create_indexes_batch('mydb', 'users', indexes)
print(results)

注意事项

# 1. 索引命名规范
def create_named_index():
    collection.create_index(
        [("order_id", 1)],
        name="idx_order_id"  # 自定义名称，便于管理
    )
# 2. 索引大小监控
def check_index_size():
    stats = db.command("collstats", "mycollection")
    total_index_size = stats.get('totalIndexSize', 0)
    index_sizes = stats.get('indexSizes', {})
    print(f"Total index size: {total_index_size} bytes")
    for index_name, size in index_sizes.items():
        print(f"Index '{index_name}': {size} bytes")
# 3. 避免过度索引
def analyze_query_patterns():
    # 分析慢查询日志
    system_profile = db.system.profile
    slow_queries = list(system_profile.find(
        {"millis": {"$gt": 100}}  # 查找执行时间超过100ms的查询
    ).sort("millis", -1).limit(10))
    return slow_queries

这些示例涵盖了MongoDB索引创建的主要场景和最佳实践,根据你的实际业务需求选择合适的索引类型和配置。