影刀RPA 数据库自动化备份MySQL PostGRESS定时快照什么情况用什么 → 怎么做 → 有什么坑作者林焱 | 飞行社出品什么情况用什么数据库每天都要备份手动执行mysqldump或pg_dump容易忘、容易出错。一旦数据丢失哭都来不及。这套方案适合企业DBA自动化数据库备份定时全量/增量备份备份文件自动上传到云存储OSS/S3核心工具影刀RPA mysqldump/pg_dump OSS/S3上传 备份验证怎么做第一步自动化MySQL备份拼多多店群自动化上架方案importosimporttimeimportsubprocessfromdatetimeimportdatetimeimportgzipdefbackup_mysql(host,port,user,password,database,backup_dir,compressTrue): 备份MySQL数据库 compress: 是否压缩备份文件推荐开启节省磁盘空间 # 1. 创建备份目录os.makedirs(backup_dir,exist_okTrue)# 2. 生成备份文件名timestampdatetime.now().strftime(%Y%m%d_%H%M%S)backup_fileos.path.join(backup_dir,f{database}_{timestamp}.sql)ifcompress:backup_file.gzprint(f开始备份MySQL数据库:{database})print(f 备份文件:{backup_file})# 3. 构造mysqldump命令cmd[mysqldump,f--host{host},f--port{port},f--user{user},f--password{password},--single-transaction,# InnoDB热备份不锁表--routines,# 备份存储过程--triggers,# 备份触发器--events,# 备份事件--hex-blob,# 二进制数据安全导出database]try:start_timetime.time()ifcompress:# 压缩备份直接生成.gz文件withgzip.open(backup_file,wb)asf:processsubprocess.Popen(cmd,stdoutf,stderrsubprocess.PIPE)_,stderrprocess.communicate()else:# 不压缩withopen(backup_file,w)asf:processsubprocess.Popen(cmd,stdoutf,stderrsubprocess.PIPE)_,stderrprocess.communicate()ifprocess.returncode!0:print(f⚠️ 备份失败:{stderr.decode(utf-8,errorsignore)})returnFalse# 4. 计算备份大小和耗时backup_size_mbos.path.getsize(backup_file)/(1024**2)elapsed_timetime.time()-start_timeprint(f✅ 备份成功)print(f 文件大小:{backup_size_mb:.2f}MB)print(f 耗时:{elapsed_time:.2f}秒)returnbackup_fileexceptExceptionase:print(f⚠️ 备份失败:{e})returnFalse# 使用示例backup_filebackup_mysql(hostlocalhost,port3306,userroot,passwordyour_password,databaseproduction_db,backup_dir/data/backups/mysql)ifbackup_file:print(f备份文件已生成:{backup_file})第二步自动化PostGRESS备份defbackup_postgress(host,port,user,password,database,backup_dir,compressTrue): 备份PostGRESS数据库 使用pg_dump逻辑备份或pg_basebackup物理备份 os.makedirs(backup_dir,exist_okTrue)timestampdatetime.now().strftime(%Y%m%d_%H%M%S)backup_fileos.path.join(backup_dir,f{database}_{timestamp}.sql)ifcompress:backup_file.gzprint(f开始备份PostGRESS数据库:{database})# 设置环境变量PGPASSWORD会被pg_dump读取envos.environ.copy()env[PGPASSWORD]password cmd[pg_dump,f--host{host},f--port{port},f--username{user},f--dbname{database},--no-owner,# 不包含OWNER语句--no-acl,# 不包含GRANT/REVOKE语句--if-exists# DROP语句加上IF EXISTS]try:start_timetime.time()ifcompress:withgzip.open(backup_file,wb)asf:processsubprocess.Popen(cmd,stdoutf,stderrsubprocess.PIPE,envenv)_,stderrprocess.communicate()else:withopen(backup_file,w)asf:processsubprocess.Popen(cmd,stdoutf,stderrsubprocess.PIPE,envenv)_,stderrprocess.communicate()ifprocess.returncode!0:print(f⚠️ 备份失败:{stderr.decode(utf-8,errorsignore)})returnFalsebackup_size_mbos.path.getsize(backup_file)/(1024**2)elapsed_timetime.time()-start_timeprint(f✅ 备份成功)print(f 文件大小:{backup_size_mb:.2f}MB)print(f 耗时:{elapsed_time:.2f}秒)returnbackup_fileexceptExceptionase:print(f⚠️ 备份失败:{e})returnFalse第三步备份文件上传到云存储importoss2# 阿里云OSSimportboto3# AWS S3defupload_to_oss(local_file,bucket_name,endpoint,access_key,secret_key,remote_pathNone): 上传备份文件到阿里云OSS try:# 1. 初始化OSS客户端authoss2.Auth(access_key,secret_key)bucketoss2.Bucket(auth,endpoint,bucket_name)# 2. 确定远程路径ifnotremote_path:remote_pathfbackups/{os.path.basename(local_file)}print(f开始上传到OSS:{remote_path})# 3. 上传文件自动断点续传oss2.defaults.max_redries3# 失败重试3次bucket.put_object_from_file(remote_path,local_file)# 4. 验证上传ifbucket.object_exists(remote_path):print(f✅ 上传成功: oss://{bucket_name}/{remote_path})returnTrueelse:print(f⚠️ 上传验证失败)returnFalseexceptExceptionase:print(f⚠️ 上传到OSS失败:{e})returnFalsedefupload_to_s3(local_file,bucket_name,aws_access_key,aws_secret_key,region,remote_pathNone): 上传备份文件到AWS S3 try:# 1. 初始化S3客户端s3boto3.client(s3,aws_access_key_idaws_access_key,aws_secret_access_keyaws_secret_key,region_nameregion)# 2. 确定远程路径ifnotremote_path:remote_pathfbackups/{os.path.basename(local_file)}print(f开始上传到S3:{remote_path})# 3. 上传文件支持断点续传fromboto3.s3.transferimportTransferConfig configTransferConfig(multipart_threshold1024*25,# 25MB以上用多分片上传max_concurrency10)s3.upload_file(local_file,bucket_name,remote_path,Configconfig)# 4. 验证上传try:s3.head_object(Bucketbucket_name,Keyremote_path)print(f✅ 上传成功: s3://{bucket_name}/{remote_path})returnTrueexcept:print(f⚠️ 上传验证失败)returnFalseexceptExceptionase:print(f⚠️ 上传到S3失败:{e})returnFalse第四步备份验证关键defverify_mysql_backup(backup_file,host,port,user,password,test_database): 验证MySQL备份文件是否可用 方法恢复到临时数据库检查表数量和数据量 print(f开始验证备份文件:{backup_file})# 1. 创建临时数据库try:importpymysql connpymysql.connect(hosthost,portport,useruser,passwordpassword)cursorconn.cursor()cursor.execute(fCREATE DATABASE IF NOT EXISTS{test_database})cursor.execute(fUSE{test_database})conn.close()exceptExceptionase:print(f⚠️ 创建临时数据库失败:{e})returnFalse# 2. 恢复备份到临时数据库try:ifbackup_file.endswith(.gz):# 解压缩并恢复cmdfgunzip {backup_file}| mysql -h{host}-P{port}-u{user}-p{password}{test_database}else:# 直接恢复cmdfmysql -h{host}-P{port}-u{user}-p{password}{test_database}{backup_file}resultsubprocess.run(cmd,shellTrue,capture_outputTrue,textTrue)ifresult.returncode!0:print(f⚠️ 恢复失败:{result.stderr})returnFalseprint(f✅ 恢复到临时数据库成功)exceptExceptionase:print(f⚠️ 恢复失败:{e})returnFalse# 3. 验证数据检查表数量和数据量try:connpymysql.connect(hosthost,portport,useruser,passwordpassword,databasetest_database)cursorconn.cursor()# 获取所有表cursor.execute(SHOW TABLES)tablescursor.fetchall()print(f 恢复表数量:{len(tables)})# 检查每个表的行数total_rows0fortableintables:table_nametable[0]cursor.execute(fSELECT COUNT(*) FROM {table_name})row_countcursor.fetchone()[0]total_rowsrow_countprint(f 总记录数:{total_rows})# 4. 删除临时数据库cursor.execute(fDROP DATABASE{test_database})conn.close()print(f✅ 备份验证通过)returnTrueexceptExceptionase:print(f⚠️ 验证失败:{e})returnFalse第五步清理旧备份文件defcleanup_old_backups(backup_dir,keep_days7,keep_count10): 清理旧备份文件 keep_days: 保留最近几天的备份 keep_count: 无论日期至少保留几个备份 print(f清理旧备份文件:{backup_dir})print(f 保留最近{keep_days}天 至少{keep_count}个文件)# 1. 获取所有备份文件backup_files[]forfilenameinos.listdir(backup_dir):iffilename.endswith(.sql)orfilename.endswith(.sql.gz):file_pathos.path.join(backup_dir,filename)mod_timeos.path.getmtime(file_path)backup_files.append((file_path,mod_time))# 2. 按修改时间排序最新的在前面backup_files.sort(keylambdax:x[1],reverseTrue)# 3. 确定要删除的文件nowtime.time()files_to_delete[]fori,(file_path,mod_time)inenumerate(backup_files):# 保留至少keep_count个文件ifikeep_count:continue# 删除超过keep_days天的文件ifnow-mod_timekeep_days*86400:files_to_delete.append(file_path)# 4. 删除文件deleted_count0deleted_size_mb0forfile_pathinfiles_to_delete:file_sizeos.path.getsize(file_path)os.remove(file_path)deleted_count1deleted_size_mbfile_size/(1024**2)print(f 删除:{os.path.basename(file_path)})print(f✅ 清理完成: 删除{deleted_count}个文件释放{deleted_size_mb:.2f}MB 空间)# 使用示例在备份完成后调用cleanup_old_backups(backup_dir/data/backups/mysql,keep_days7,# 保留最近7天keep_count10# 至少保留10个备份)第六步影刀RPA完整流程编排【定时触发】每天凌晨2:00业务低峰期 ↓ 【Python节点】backup_mysql() → 备份MySQL ↓ 【条件判断】备份是否成功 ├─ 否 → 【企微告警】发送备份失败通知 └─ 是 → 继续 ↓ 【Python节点】verify_mysql_backup() → 验证备份 ↓ 【条件判断】验证是否通过 ├─ 否 → 【企微告警】发送验证失败通知 └─ 是 → 继续 ↓ 【Python节点】upload_to_oss() / upload_to_s3() → 上传到云存储 ↓ 【条件判断】上传是否成功 ├─ 否 → 【企微告警】发送上传失败通知 └─ 是 → 继续 ↓ 【Python节点】cleanup_old_backups() → 清理旧备份 ↓ 【写入数据库】记录备份日志时间、大小、路径 ↓ 【生成报告】数据库备份日报.xlsx → 包含备份时间、大小、验证结果、上传状态 ↓ 【发送邮件】将报告发送给DBA团队有什么坑坑1备份时锁表影响业务使用mysqldump备份MyISAM表时会锁表导致业务写入失败。解决方案使用--single-transaction参数InnoDB支持热备份不锁表# 正确做法InnoDB表使用--single-transactioncmd[mysqldump,--single-transaction,# InnoDB热备份不锁表# ... 其他参数]# 如果是MyISAM表只能在业务低峰期备份或者改用--lock-tables坑2备份文件过大传输超时几百GB的备份文件上传到云存储时可能超时失败。解决方案使用断点续传和多分片上传# OSS断点续传importoss2 bucketoss2.Bucket(auth,endpoint,bucket_name)# 开启断点续传oss2.defaults.max_redries3bucket.put_object_from_file(remote_path,local_file)# S3多分片上传自动断点续传[video(video-lMdNWGfo-1784265027648)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营)]fromboto3.s3.transferimportTransferConfig configTransferConfig(multipart_threshold1024*25,# 25MB以上用多分片上传max_concurrency10# 最多10个并发线程)s3.upload_file(local_file,bucket_name,remote_path,Configconfig)坑3备份验证占用大量磁盘空间验证备份时需要恢复到临时数据库可能占用和备份文件一样大的磁盘空间。解决方案使用单独的验证服务器或者只验证表结构而不验证数据defverify_backup_light(backup_file,host,port,user,password,test_database):轻量级验证只检查表结构# 1. 恢复备份只恢复表结构不恢复数据cmdfmysql -h{host}-P{port}-u{user}-p{password}{test_database}{backup_file}# 如果是大数据库可以只恢复表结构在备份文件中查找CREATE TABLE语句# 2. 只检查表数量不检查数据量# ...returnTrue坑4云存储成本失控每天全量备份上传到云存储一个月后账单吓死人。解决方案使用增量备份 生命周期策略# 1. 全量备份每周一次ifdatetime.now().weekday()0:# 周一全量备份backup_filebackup_mysql(...,fullTrue)# 2. 增量备份其他天else:backup_filebackup_mysql_incremental(...)# 3. OSS生命周期策略自动删除旧备份# 在OSS控制台设置30天后自动删除总结功能节省时间附加价值自动备份每天省30分钟避免人为遗忘自动上传每天省15分钟异地容灾更安全自动验证—确保备份可用避免备份无效自动清理每周省30分钟节省磁盘空间实际落地建议先在测试环境验证确保备份和恢复流程正确使用_crontab_或影刀定时触发每天低峰期自动执行做好权限管理备份文件权限设置为600只有owner可读定期恢复演练每月做一次完整的恢复演练验证备份真的可用数据库自动化备份能为DBA团队节省80%以上的日常备份时间同时提高数据安全性避免因人为疏忽导致的数据丢失。
