批量清理mongodb歷史數(shù)據(jù)
清理程序的原來
- 目前項(xiàng)目組上很多平臺(tái)上線歷史數(shù)據(jù)積壓懒浮，導(dǎo)致入庫查詢數(shù)據(jù)緩慢飘弧，歷史數(shù)據(jù)有些已經(jīng)歸檔，進(jìn)行歷史數(shù)據(jù)清理刪除砚著。
- 之前臨時(shí)寫shell腳本次伶，太簡陋，重新使用Python進(jìn)行改造稽穆，新增備份功能冠王，和配置文件刪除指定字段和時(shí)間范圍內(nèi)數(shù)據(jù)。
代碼篇
#!/usr/local/python3/bin/python3
import configparser,logging.config,sys,os,subprocess
import pymongo,ast
# from pymongo import MongoClient
from datetime import datetime,timedelta
from urllib import parse
def init_mongodb(MongoDBAuth):
if mongodb_auth:
username = parse.quote_plus(mongodb_user)
password = parse.quote_plus(mongodb_passwd)
ConnPasswd = "mongodb://" + username + ":" + password + "@" + mongodb_ip + ":" + mongodb_port + "/"
try:
clients = pymongo.MongoClient(ConnPasswd)
logger.info("init mongodb conn: " + ConnPasswd)
return clients
except Exception as e:
logger.info("use mongodb user pass conn err: " + str(e))
return False
else:
try:
clients = pymongo.MongoClient(mongodb_ip, int(mongodb_port))
logger.info("init mongodb conn: " + mongodb_ip +":" +mongodb_port)
return clients
except Exception as e:
logger.info("use mongodb user pass conn err: " + str(e))
return False
#查看出全部db
def get_mongodb_dbname():
db_names_list = []
db_names = mongo_client.list_database_names()
for db_name in db_names:
db_names_list.append(db_name)
for filter_dbname in need_filter_dbname_list:
if filter_dbname in db_names_list:
db_names_list.remove(filter_dbname)
logger.info("delete need filter dbname: " + filter_dbname)
# logger.info("get all db_name: " +str(db_names_list))
return db_names_list
#查詢出db中全部表
def get_mongodb_tables(entid):
db_collections_list = []
db=mongo_client[entid]
collections = db.list_collection_names()
for collection in collections:
db_collections_list.append(collection)
logger.debug("get " + entid + " all collections: " +str(db_collections_list))
return db_collections_list
#查詢集合中索引索引和是否分片
def get_index_key_tables(entid,collection_name):
index_list = []
formatted_results = []
db=mongo_client[entid]
collection=db[collection_name]
indexes = collection.list_indexes()
ns_name = entid + "." + collection_name
for result in indexes:
formatted_result = {k.upper(): v for k, v in result.items()}
each_key = formatted_result.get("KEY")
ns_name = formatted_result.get("NS")
ok_index = {key: value for key, value in each_key.items()}
index_list.append(ok_index)
index_list = result = [d for d in index_list if not (isinstance(d, dict) and '_id' in d and d['_id'] == 1)]
collection_stats = db.command("collstats", collection_name)
collection_sharded = collection_stats.get("sharded", False)
if len(index_list) != 0:
logger.debug("get collection " + ns_name + " index: " +str(index_list))
#logger.info("get now In the collection " + ns_name + " sharded status: " +str(collection_sharded))
return index_list,collection_sharded
#創(chuàng)建集合索引
def craete_index(entid,collection_name,index):
db=mongo_client[entid]
collection=db[collection_name]
logger.info("need craete index: " + entid +"."+collection_name + " : "+ str(index))
# index = (list(index.keys())[0], list(index.values())[0])
index = [(k, v) for k, v in index.items()]
result = collection.create_index(index)
logger.info("mongodb " +entid +"."+collection_name + " create index return msg: " + str(result) )
#查看對(duì)應(yīng)dbname是否已經(jīng)是shards,棄用
def is_database_sharded(database_name):
db = mongo_client["admin"]
sharded_databases = db.command("listshards")["shards"]
for shard in sharded_databases:
if database_name in db.command("listdatabases")["databases"]:
return True
return False
#創(chuàng)建分片索引片鍵
def create_sharded_func(entid, collection_name, shard_key):
db = mongo_client["admin"]
collection_path = '{}.{}'.format(entid, collection_name)
logger.info("need craete sharded key : " + collection_path + " : " + str(shard_key))
sharding_colunm,sharding_type = "",""
for key, value in shard_key.items():
sharding_colunm= key
sharding_type = value
try:
db.command('enableSharding', entid)
except Exception as e:
logger.error("create dbname sharded key error: return: " + str(e))
try:
result = db.command('shardCollection', collection_path,key = {sharding_colunm:sharding_type})
logger.info(entid + "." + collection_path + " create sharded key return: " + str(result))
except Exception as e:
logger.error("create sharded key error: return: " + str(e))
#讀取文件獲取對(duì)應(yīng)索引和片鍵key信息
def read_file_index(index_file):
index_list = []
Shard_list = []
with open(index_file, 'r') as f:
for line in f.readlines():
line = line.replace(" ", "")
#通過mongodbShard: 來區(qū)分那個(gè)片鍵的可以舌镶，寫
# print(line)
if "mongodbShard:" not in line:
table, key_str = line.strip().split("=")
key = ast.literal_eval(key_str)
index_list.append({table: key})
else:
Shard_key_str = line.strip().split("mongodbShard:")[1]
Shard_key_str = ast.literal_eval(Shard_key_str)
Shard_list.append(Shard_key_str)
return index_list,Shard_list
#獲取多少天前的時(shí)間戳
def get_timestamp_days_ago(get_days):
# 獲取當(dāng)前日期和時(shí)間
now = datetime.now()
# 減去30天
date_30_days_ago = now - timedelta(days=int(get_days))
# 將結(jié)果轉(zhuǎn)換為當(dāng)天的整點(diǎn)00:00:00
date_start_of_day = date_30_days_ago.replace(hour=0, minute=0, second=0, microsecond=0)
# 將結(jié)果轉(zhuǎn)換為時(shí)間戳
timestamp = int(date_start_of_day .timestamp())
return timestamp
#判斷字符串類型和長度對(duì)應(yīng)返回需要?jiǎng)h除的時(shí)間字段值
def if_string_type(data_stamp):
del_timestamp = ""
get_need_del_timestamp = get_timestamp_days_ago(int(Del_day))
if isinstance(data_stamp, str) and len(data_stamp) == 10:
del_timestamp = str(get_need_del_timestamp)
if isinstance(data_stamp, str) and len(data_stamp) == 13:
del_timestamp = str(get_need_del_timestamp) + "000"
if isinstance(data_stamp, int) and len(str(data_stamp)) == 10:
del_timestamp = get_need_del_timestamp
if isinstance(data_stamp, int) and len(str(data_stamp)) == 13:
del_timestamp = int(get_need_del_timestamp) * 1000
return del_timestamp
#獲取該集合中一條數(shù)據(jù)
def get_one_data(entid,collection_name):
db=mongo_client[entid]
collection=db[collection_name]
Filter_conditions_key = str(need_del_table_field)
result = collection.find_one({}, {**{Filter_conditions_key: 1}, '_id': 0})
if result and Filter_conditions_key in result:
start_time_value = result.get(Filter_conditions_key)
logger.debug("get "+ entid + "." + collection_name + " Corresponding " +Filter_conditions_key + " field value: " + str(start_time_value) )
return start_time_value
else:
# logger.info("No " +Filter_conditions_key + " field found in the document. return: " + str(result) )
return False
# 按照日期刪除該集合中歷史數(shù)據(jù)
def del_data(entid,collection_name,get_del_timestamp):
db=mongo_client[entid]
collection=db[collection_name]
Filter_conditions_key = str(need_del_table_field)
Filter_conditions_value = get_del_timestamp
try:
result = collection.delete_many({Filter_conditions_key: {"$lt": Filter_conditions_value}})
logger.info(entid +" run sql: db"+"."+collection_name+".remove({"+Filter_conditions_key+ ":"+"{$lt:"+str(Filter_conditions_value) +"})")
if result.deleted_count > 0:
logger.info("By date delete " + str(entid) + "." + collection_name + " less than " + str(get_del_timestamp) + " del document count: " + str(result.deleted_count))
except Exception as e:
logger.error("Error occurred while deleting documents: " + str(e))
# 刪除該集合中全部歷史數(shù)據(jù)
def del_all_data(entid,collection_name):
db=mongo_client[entid]
collection=db[collection_name]
try:
result = collection.delete_many({})
if result.deleted_count > 0:
logger.info(entid + " run sql: db"+"."+collection_name+".remove({})")
logger.info(entid + "." + collection_name + " del all document count: " + str(result.deleted_count))
except Exception as e:
logger.info(entid + "." + collection_name + " del all document error: " + str(result) )
# 備份數(shù)據(jù)
def dump_mongodb_data(dbname,table,not_quiet_dump,del_time):
status_info = ["1"]
if is_del_bakcup_data:
if os.path.exists(mongodump_command_path):
run_status = " && echo $?"
run_commnd = ""
if not_quiet_dump:
if mongodb_auth:
#run_commnd = mongodump_command_path + " -h " + mongodb_ip + ":" + str(mongodb_port) + " --authenticationDatabase=" +mongodb_auth_db + " -u " + mongodb_user + " -p " + mongodb_passwd + " -d " + dbname + " -c " + table + " -q '{" + need_del_table_field + ": {" + + "}}'" + " -o " + bakcup_dir_path
run_command = f"{mongodump_command_path} -h {mongodb_ip}:{mongodb_port} --authenticationDatabase={mongodb_auth_db} -u {mongodb_user} -p {mongodb_passwd} -d {dbname} -c {table} -q '{{\"{need_del_table_field}\": {{\"$lt\": \"{del_time}\"}}}}' -o {bakcup_dir_path}"
else:
# run_commnd = mongodump_command_path + " -h " + mongodb_ip + ":" + str(mongodb_port) + " -d " + dbname + " -c " + table + " -o " + bakcup_dir_path
run_commnd = f"{mongodump_command_path} -h {mongodb_ip}:{mongodb_port} -d {dbname} -c {table} -q '{{\"{need_del_table_field}\": {{\"$lt\": \"{del_time}\"}}}}' -o {bakcup_dir_path}"
else:
if mongodb_auth:
# run_commnd = mongodump_command_path + " -h " + mongodb_ip + ":" + str(mongodb_port) + " --authenticationDatabase=" +mongodb_auth_db + " -u " + mongodb_user + " -p " + mongodb_passwd + " -d " + dbname + " -c " + table + " -o " + bakcup_dir_path
run_command = f"{mongodump_command_path} -h {mongodb_ip}:{mongodb_port} --authenticationDatabase={mongodb_auth_db} -u {mongodb_user} -p {mongodb_passwd} -d {dbname} -c {table} -o {bakcup_dir_path}"
else:
# run_commnd = mongodump_command_path + " -h " + mongodb_ip + ":" + str(mongodb_port) + " -d " + dbname + " -c " + table + " -o " + bakcup_dir_path
run_commnd = f"{mongodump_command_path} -h {mongodb_ip}:{mongodb_port} -d {dbname} -c {table} -o {bakcup_dir_path}"
logger.info("run command: " + run_commnd)
try:
msg = os.popen(run_commnd + run_status)
status_info = [line.strip() for line in msg.readlines()]
logger.info("mongodump command result: " + str(status_info))
except Exception as e:
logger.error("mongodump command error: " + str(e))
else:
logger.info("mongodump command file not exists ," + mongodump_command_path)
else:
logger.debug("config file not set is_del_bakcup_data = True, not dump data")
return status_info
if __name__=="__main__":
cfgpath = "./cfg/config.ini"
conf = configparser.ConfigParser()
conf.read(cfgpath)
mongodb_ip = conf.get("main", "mongodb_ip")
mongodb_port = conf.get("main", "mongodb_port")
mongodb_auth = conf.getboolean("main", "mongodb_auth")
mongodb_user = conf.get("main", "mongodb_user")
mongodb_passwd = conf.get("main", "mongodb_passwd")
mongodb_auth_db = conf.get("main", "mongodb_auth_db")
need_filter_dbname = conf.get("main", "need_filter_dbname")
is_del_bakcup_data = conf.getboolean("main", "is_del_bakcup_data")
bakcup_dir_path = conf.get("main", "bakcup_dir_path")
mongodump_command_path = conf.get("main", "mongodump_command_path")
Del_day = conf.get("main", "Del_day")
need_del_table_field = conf.get("main", "need_del_table_field")
need_del_table_list = conf.get("main", "need_del_table_list")
need_del_table_list = [item for item in need_del_table_list.split(",") if item != '']
need_del_null_table_list = conf.get("main", "need_del_null_table_list")
need_del_null_table_list = [item for item in need_del_null_table_list.split(",") if item != '']
auth_get_entid = conf.getboolean("main", "auth_get_entid")
need_filter_dbname_list = [item for item in need_filter_dbname.split(",") if item != '']
#獲取配置項(xiàng)
all_ent_id = conf.get("main", "ent_id")
get_dbname_list = all_ent_id.split(",")
logging.config.fileConfig("./cfg/logger.conf")
logger = logging.getLogger("rotatfile")
# 初始化 MongoDB
mongo_client = init_mongodb(mongodb_auth)
if mongo_client:
logger.info("MongoDB init successfully")
else:
logger.error("Failed to initialize MongoDB")
sys.exit(10)
if auth_get_entid:
get_dbname_list = get_mongodb_dbname()
logger.info("get all dbname list: " + str(get_dbname_list))
else:
logger.info("file get dbname list: " + str(get_dbname_list))
for dbname in get_dbname_list:
get_end_all_table = get_mongodb_tables(dbname)
for table in need_del_table_list:
get_one_data_mes = get_one_data(dbname,table)
if table in get_end_all_table:
get_index_key_tables(dbname,table)
else:
logger.error(dbname + " not have table: " + table)
continue
# break
#刪除按照日期數(shù)據(jù)
if get_one_data_mes:
get_del_timestmap = if_string_type(get_one_data_mes)
if dump_mongodb_data(dbname,table,True,get_del_timestmap)[0] == '0' or is_del_bakcup_data == False:
if get_del_timestmap:
del_data(dbname,table,get_del_timestmap)
else:
logger.error("get del timestmap fail")
else:
if is_del_bakcup_data == False:
logger.error("is_del_bakcup_data seting False, dump mongodb data fail")
else:
logger.error("dump mongodb data fail, but is del backup data")
for null_table in need_del_null_table_list:
if dump_mongodb_data(dbname,null_table,False,"1")[0] == '0' or is_del_bakcup_data == False:
if null_table in get_end_all_table:
#刪除全部歷史數(shù)據(jù)
del_all_data(dbname,null_table)
else:
logger.error( dbname + " not have table: " + null_table)
else:
if is_del_bakcup_data == False:
logger.error("is_del_bakcup_data seting False, dump mongodb data fail")
else:
logger.error("dump mongodb data fail, but is del backup data")
mongo_client.close()
logger.info("MongoDB closed")
配置文件篇
- 該配置項(xiàng)大概使用說明
- 支持刪除指定時(shí)間前柱彻，進(jìn)行數(shù)據(jù)備份在刪除豪娜，根據(jù)不同配置項(xiàng)進(jìn)行配置；
- 同理可支持不進(jìn)行備份绒疗，也可以清理刪除侵歇，根據(jù)不同配置項(xiàng)進(jìn)行配置；
- 根據(jù)字段來查詢過濾吓蘑。
[DEFAULT]
mongodb_ip = 10.130.47.197
mongodb_port = 40000
mongodb_auth = False
mongodb_user = admin
mongodb_passwd = test@123
mongodb_auth_db = admin
#從全部dbname中進(jìn)行過濾不需要處理的dbname，使用逗號(hào)分割
need_filter_dbname = local,config,admin
#指定需要按照日期刪除的集合坟冲，使用逗號(hào)分割
need_del_table_list = new_r_ags_e_back,call_detail_back
#指定需要按照日期刪除的集合字段過濾
need_del_table_field = start_time
#指定清空刪除的集合磨镶，使用逗號(hào)分割
need_del_null_table_list = call_duration_cache,duration_cache
[main]
#是否自動(dòng)獲取對(duì)應(yīng)mongodb中全部dbname
auth_get_entid = False
#從配置文件中獲取dbname
ent_id = 20241205,20250107
#需要?jiǎng)h除多少天以前的數(shù)據(jù)
Del_day = 97
#是否需要備份數(shù)據(jù)
is_del_bakcup_data = False
#備份目錄
bakcup_dir_path = ./data
#備份命令路徑
mongodump_command_path = /home/devops/Python/Mongodb_del_history/mongodump
腳本運(yùn)行
- 腳本運(yùn)行
[devops@db1 Mongodb_del_history]$ tar xf Mongodb_del_history.tar.gz
[devops@db1 Mongodb_del_history]$ cd Mongodb_del_history
[devops@db1 Mongodb_del_history]$ nohup ./del_history_data &
2025-01-06 14:15:01 139749303605056 del_history_data.py:24 INFO init mongodb conn: 10.130.47.197:40000
2025-01-06 14:15:01 139749303605056 del_history_data.py:303 INFO MongoDB init successfully
2025-01-06 14:15:01 139749303605056 del_history_data.py:39 INFO delete need filter dbname: local
2025-01-06 14:15:01 139749303605056 del_history_data.py:310 INFO get all dbname list: ['0103290010', '0103290012', '0103290013', '0103290015']
2025-01-06 14:15:01 139749303605056 del_history_data.py:321 ERROR 0103290010 not have table: jhk_task_status
2025-01-06 14:15:01 139749303605056 del_history_data.py:321 ERROR 0103290010 not have table: sd_call_detail_back
2025-01-06 14:15:01 139749303605056 del_history_data.py:229 INFO run command: /home/devops/Python/Mongodb_del_history/mongodump -h 10.130.47.197:40000 -d 0103290010 -c call_duration_cache -o ./data
2025-01-06 14:15:01 139749303605056 del_history_data.py:233 INFO mongodump command result: ['0']
2025-01-06 14:15:01 139749303605056 del_history_data.py:229 INFO run command: /home/devops/Python/Mongodb_del_history/mongodump -h 10.130.47.197:40000 -d 0103290010 -c duration_cache -o ./data
2025-01-06 14:15:01 139749303605056 del_history_data.py:233 INFO mongodump command result: ['0']
2025-01-06 14:15:01 139749303605056 del_history_data.py:347 INFO MongoDB closed
二進(jìn)制文件程序下載
- 使用鏈接下載
wget https://zhao138969.com/LinuxPackage/Python/del_history_data
本文由mdnice多平臺(tái)發(fā)布
