影刀RPA 数据备份自动化增量备份与恢复演练什么情况用什么 → 怎么做 → 有什么坑作者林焱 | 飞行社出品什么情况用什么每天全量备份数据库文件越来越大备份时间越来越长最后备份窗口都兜不住了。这套方案适合企业DBA自动化增量备份定时全量增量组合备份自动恢复演练验证备份真的能恢复核心工具影刀RPA MySQL binlog/PostgreSQL WAL 恢复验证脚本怎么做拼多多店群自动化报活动上架第一步MySQL增量备份binlogimportosimportsubprocessfromdatetimeimportdatetimedefmysql_incremental_backup(host,port,user,password,backup_dir): MySQL增量备份通过复制binlog实现 前提MySQL已开启binlogmy.cnf中log_binON os.makedirs(backup_dir,exist_okTrue)# 1. 查看当前binlog文件cmdfmysql -h{host}-P{port}-u{user}-p{password}-e SHOW MASTER STATUSresultsubprocess.run(cmd,shellTrue,capture_outputTrue,textTrue)ifresult.returncode!0:print(f⚠️ 获取binlog位置失败:{result.stderr})returnFalse# 解析当前binlog文件和位置linesresult.stdout.strip().split(\n)iflen(lines)2:print(⚠️ 无法解析binlog状态)returnFalsecurrent_loglines[1].split(\t)[0]current_poslines[1].split(\t)[1]# 2. 复制新的binlog文件到备份目录# 找到所有新的binlog文件比上次备份新的binlog_dir/var/lib/mysql# binlog文件目录根据实际修改binlog_prefixmysql-bin# binlog前缀根据实际修改copied0forfnameinsorted(os.listdir(binlog_dir)):ifnotfname.startswith(binlog_prefix):continuesrcos.path.join(binlog_dir,fname)dstos.path.join(backup_dir,fname)ifnotos.path.exists(dst)oros.path.getmtime(src)os.path.getmtime(dst):importshutil shutil.copy2(src,dst)copied1print(f✅ 增量备份完成新增{copied}个binlog文件)print(f 当前位置:{current_log}{current_pos})returnTrue# 使用示例mysql_incremental_backup(hostlocalhost,port3306,userbackup_user,passwordsecret,backup_dir/data/backup/mysql/incremental)第二步PostgreSQL增量备份WAL归档defpostgres_incremental_backup(pg_data_dir,backup_dir): PostgreSQL增量备份WAL文件归档 前提已配置wal_levelreplica, archive_modeon os.makedirs(backup_dir,exist_okTrue)wal_diros.path.join(pg_data_dir,pg_wal)ifnotos.path.exists(wal_dir):print(f⚠️ WAL目录不存在:{wal_dir})returnFalsecopied0forfnameinos.listdir(wal_dir):iflen(fname)!24:# WAL文件名长度固定continuesrcos.path.join(wal_dir,fname)dstos.path.join(backup_dir,fname)ifnotos.path.exists(dst):importshutil shutil.copy2(src,dst)copied1print(f✅ PostgreSQL WAL归档备份完成新增{copied}个WAL文件)returnTrue# 配合postgresql.conf配置# wal_level replica# archive_mode on# archive_command cp %p /data/backup/pg/incremental/%f第三步自动恢复演练关键defverify_mysql_recovery(backup_dir,target_db,host,port,user,password): 自动恢复验证将备份恢复到临时实例验证数据完整性 这是最重要的步骤很多人的备份从来没验证过能不能恢复。 importtempfileimportshutil# 1. 创建临时恢复目录withtempfile.TemporaryDirectory()astmpdir:print(f开始恢复验证临时目录:{tmpdir})# 2. 恢复全量备份full_backupfind_latest_full_backup(backup_dir)ifnotfull_backup:print(⚠️ 找不到全量备份文件)returnFalseprint(f恢复全量备份:{full_backup})cmdfmysql -h{host}-P{port}-u{user}-p{password}{target_db}{full_backup}resultsubprocess.run(cmd,shellTrue,capture_outputTrue,textTrue)ifresult.returncode!0:print(f⚠️ 全量恢复失败:{result.stderr})returnFalse# 3. 应用增量备份binlogbinlog_filessorted([fforfinos.listdir(backup_dir)iff.startswith(mysql-bin)])forbinloginbinlog_files:binlog_pathos.path.join(backup_dir,binlog)cmdfmysqlbinlog{binlog_path}| mysql -h{host}-P{port}-u{user}-p{password}{target_db}resultsubprocess.run(cmd,shellTrue,capture_outputTrue,textTrue)ifresult.returncode!0:print(f⚠️ 应用binlog失败:{binlog})returnFalseprint(✅ 增量恢复完成)# 4. 验证数据完整性validation_queries[SELECT COUNT(*) FROM users,SELECT COUNT(*) FROM orders,SELECT MAX(created_at) FROM orders]forqueryinvalidation_queries:cmdfmysql -h{host}-P{port}-u{user}-p{password}{target_db}-e \{query}\resultsubprocess.run(cmd,shellTrue,capture_outputTrue,textTrue)ifresult.returncode!0:print(f⚠️ 验证查询失败:{query})returnFalseprint(f 验证:{query}-{result.stdout.strip()})print(✅ 恢复验证通过)returnTruedeffind_latest_full_backup(backup_dir):查找最新的全量备份文件files[fforfinos.listdir(backup_dir)iff.endswith(.sql.gz)orf.endswith(.sql)]ifnotfiles:returnNonefiles.sort(keylambdaf:os.path.getmtime(os.path.join(backup_dir,f)),reverseTrue)returnos.path.join(backup_dir,files[0])第四步影刀RPA编排完整备份流程【启动】每天凌晨2:00业务低峰 ↓ 【Python节点】mysql_full_backup() → 每周日做全量备份 ↓ 【条件判断】今天是周日 ├─ 是 → 【Python节点】mysql_full_backup() └─ 否 → 【Python节点】mysql_incremental_backup() → 增量备份 ↓ 【Python节点】upload_to_oss() → 上传到云存储 ↓ 【条件判断】周六→ 做恢复演练 ├─ 是 → 【Python节点】verify_mysql_recovery() → 恢复验证 │ ↓ │ 【条件判断】验证通过 │ ├─ 是 → 【企微通知】✅ 本周备份恢复演练通过 │ └─ 否 → 【企微告警】 备份恢复失败需要人工介入 └─ 否 → 跳过演练 ↓ 【Python节点】cleanup_old_backups() → 清理30天前的备份 ↓ 【写入Excel】更新备份执行日志.xlsx有什么坑坑1增量备份依赖全量备份如果全量备份损坏所有增量备份都废了。全量备份必须定期验证可恢复性。解决方案全量备份每周至少1次每次全量备份后自动做恢复验证3-2-1原则3份副本、2种介质、1个异地备份坑2binlog/WAL文件丢失数据库崩溃时最新的binlog可能还没归档导致数据丢失。解决方案# 备份前先刷新binlogdefflush_mysql_logs(host,port,user,password):刷新binlog确保当前内存中的日志写入磁盘cmdfmysql -h{host}-P{port}-u{user}-p{password}-e FLUSH LOGSsubprocess.run(cmd,shellTrue)print(✅ binlog已刷新到磁盘)坑3恢复验证消耗大量资源TEMU店群矩阵自动化运营核价报活动在Production服务器上做恢复验证会和真实业务争抢资源。解决方案用专门的备用服务器做恢复验证或者容器化# 用Docker启动临时MySQL实例做恢复验证dockerrun-d--namemysql-restore-eMYSQL_ROOT_PASSWORDtest\-v/data/backup:/backup mysql:8.0# 在容器内恢复dockerexecmysql-restore mysql-uroot-ptest/backup/full_backup.sql坑4云存储上传失败没发现OSS/S3上传失败但备份脚本没报错等到要恢复时才发现备份文件损坏。解决方案上传后做checksum校验importhashlibdefverify_upload(local_file,oss_bucket,oss_key):上传后验证文件完整性# 本地计算MD5defcalc_md5(filepath):hash_md5hashlib.md5()withopen(filepath,rb)asf:forchunkiniter(lambda:f.read(4096),b):hash_md5.update(chunk)returnhash_md5.hexdigest()local_md5calc_md5(local_file)# 获取OSS文件ETag通常是MD5importoss2 authoss2.Auth(access_key,secret_key)bucketoss2.Bucket(auth,endpoint,bucket_name)headbucket.head_object(oss_key)oss_md5head.headers[ETag].strip()iflocal_md5oss_md5:print(f✅ 上传验证通过:{local_md5})returnTrueelse:print(f 上传验证失败本地{local_md5}OSS{oss_md5})returnFalse总结备份类型备份速度恢复速度存储空间推荐频率全量备份慢快大每周1次增量备份快慢需逐层应用小每天1次差异备份中中中每天1次最重要的经验备份不做恢复验证 没备份。一定要定期建议每周自动做恢复演练并通知结果。