MongoDB数据库（10亿条数据）清理策略: 自动化过期数据删除实战

1、引言

随着应用程序和业务数据的持续增长，有效地管理数据库存储空间成为维护系统性能的关键。在MongoDB这类NoSQL数据库中，定期清理过期数据变得尤为重要，这不仅能释放宝贵的存储资源，还能优化查询性能，确保数据库运行的高效与稳定。
本文将深入探讨一种自动化清理MongoDB中过期数据的策略，并通过一个实际的Python脚本示例，展示如何实现这一功能。

2、需求背景

根据公司业务发展积累，在众多应用场景中，如日志记录、临时缓存、会话管理等，数据往往具有时效性，超过一定时间后便不再有用。如果不及时清理，这些过期数据会占用大量存储空间，增加数据库维护成本，甚至影响查询效率。
目前我们的 MongoDB数据库单表达到70G，冗余数据积累。导致空间占用极大。为了实现“降本增效” 清理过期的数据（切忌：过期数据也需要使用mongodump备份）因此，我们需要一个自动化机制，能够根据数据的“最后修改日期”等时间戳字段，识别并删除过期记录。

3、功能概述

本方案设计了一个Python脚本，集成了以下几个核心功能：

配置文件读取：允许用户灵活配置数据库连接信息、目标集合名、数据过期天数以及批处理大小等参数。
动态时间阈值计算：根据用户设定的过期天数，计算出需删除数据的截止时间戳。
分批删除机制：为了减少对数据库的冲击，脚本采用分批删除策略，每次只处理一批数据，直至所有过期数据被清理完毕。
进度可视化：集成tqdm库，实时显示删除进度，使操作过程透明且直观。
错误处理：包含了对配置加载、数据库连接、数据操作等环节的异常处理，确保脚本的健壮性。

4、实现步骤

1、数据库表结构分析

假如我们有个：tag_logs 的集合
数据格式如下：

db.getCollection("tag_logs").insert( {_id: ObjectId("65dd5f067db3e415f0d3972f"),taskId: "65dd5efd7db3e415f0d39630",modelId: "6285a9890d45000030004392",name: "nihaogengx",ruleResult: "NOT_HIT",logic: "AND",conditionResults: [{name: "nihaogengx",result: "NOT_HIT",logic: "AND",subRuleResults: [{name: "nihaogengx",result: "NOT_HIT",variableCode: "var-instant-core-xxxxxx"}]}],type: "AUDIT_TAG",createdDate: NumberLong("1709006598851"),lastModifiedDate: NumberLong("1709006598851"),_class: "com.fujfu.shinji.entity.TagResultDO"
} );

索引查询

db.createCollection("tag_logs");db.getCollection("tag_logs").createIndex({taskId: NumberInt("1")
}, {name: "idx_tagResult_taskId"
});db.getCollection("tag_logs").createIndex({createdDate: NumberInt("1")
}, {name: "createdDate_1",background: true
});db.getCollection("tag_logs").createIndex({lastModifiedDate: NumberInt("-1")
}, {name: "lastModifiedDate_-1",background: true
});

2、增加索引

我们是根据 lastModifiedDate 来获取过期的时间，所以这个必选加索引。如果没有索引，根据下方添加

db.tag_logs.createIndex( { lastModifiedDate: -1 }, { background: true } )

这个命令的作用是在 tag_logs 集合上创建一个索引。具体来说：

db.tag_logs.createIndex：这是在 tag_logs 集合上创建索引的方法。
{ lastModifiedDate: -1 }：这是索引的键和排序顺序。具体解释如下：
- lastModifiedDate 是你希望创建索引的字段名。
- -1 表示你希望按照该字段的降序排序来创建索引。如果你用的是 1，则表示按照升序排序。
{ background: true }：这是索引创建的选项。具体解释如下：
- background: true 表示在后台创建索引。这意味着索引创建操作不会阻塞其他数据库操作，允许其他读写操作继续进行。这对于生产环境中的大型集合非常有用，因为它可以减少对应用程序正常操作的干扰。

3、脚本核心逻辑

config.ini

[database]
uri = mongodb://root:xxxx.88@mongo2.fat.xxxx.fjf:27017/?authSource=admin  #Mongo连接字符串
db_name = xxx-xxx-engine   # 数据库名称
collection_name = variable_result_1  # 集合名称
expired_days = 90  # 删除过期多少天的。 删除3个月之前的数据
batch_size=1000 #每次删除的条数

clean_expired_data.py

import configparser
from pymongo import MongoClient, errors
from datetime import datetime, timedeltafrom tqdm import tqdmdef load_config(file_path='config.ini'):"""Load configuration from the specified file."""config = configparser.ConfigParser()config.read(file_path)return configdef get_mongo_client(uri):"""Create and return a MongoDB client."""return MongoClient(uri)def get_cutoff_timestamp(days):"""Calculate and return the cutoff timestamp."""cutoff_date = datetime.now() - timedelta(days=days)return int(cutoff_date.timestamp() * 1000)def delete_expired_documents(collection, cutoff_timestamp, batch_size):"""Delete documents older than the cutoff timestamp in batches."""total_deleted = 0all_documents = collection.count_documents({})# 1. 查询出需要删除的集合数量total_to_delete = collection.count_documents({'lastModifiedDate': {'$lt': cutoff_timestamp}})print(f"集合总数: {all_documents}, 需要删除的文档数量: {total_to_delete}")# 2. 使用 tqdm 显示进度条with tqdm(total=total_to_delete, desc='Deleting documents', unit='doc') as pbar:while True:documents = collection.find({'lastModifiedDate': {'$lt': cutoff_timestamp}},limit=batch_size)document_ids = [doc['_id'] for doc in documents]if not document_ids:breakresult = collection.delete_many({'_id': {'$in': document_ids}})deleted_count = result.deleted_counttotal_deleted += deleted_count# print(f'Deleted {deleted_count} documents')# 3. 更新进度条pbar.update(deleted_count)if deleted_count < batch_size:breakreturn total_deleteddef clean_mongo_expired_data():"""Main function to clean expired data from MongoDB."""config = load_config()try:uri = config['database']['uri']db_name = config['database']['db_name']collection_name = config['database']['collection_name']expired_days = int(config['database']['expired_days'])batch_size = int(config['database']['batch_size'])client = get_mongo_client(uri)db = client[db_name]collection = db[collection_name]cutoff_timestamp = get_cutoff_timestamp(expired_days)total_deleted = delete_expired_documents(collection, cutoff_timestamp, batch_size)print('Completed deletion')print(f'Deleted {total_deleted} documents')except (configparser.Error, ValueError, errors.PyMongoError) as e:print(f'Error occurred: {e}')if __name__ == '__main__':clean_mongo_expired_data()

requirements.txt
python 环境版本：Python 3.8.10

pymongo==4.3.3
tqdm==4.66.4

5、实战测试

python3  -m venv py3  #创建虚拟环境source env_py/py3/bin/activate #加载环境pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 安装依赖更改config.ini 启动程序nohup python clean_expired_data.py  &(py3) [root@jenkins mongodb_clean]# tail -f nohup.out 集合总数: 410565470, 需要删除的文档数量: 404724244
Deleting documents:  13%|█▎        | 53910000/404724244 [1:17:54<8:13:39, 11844.06doc/s]