mysql binlog 数据恢复 记一次事故数据恢复

引言

最近项目即将上线,收到任务先行在正式环境上备份还原库;于是有了以下命令:

mysqldump -u root -p xxxx > xxxx20241012.sql

使用标准mysqldump备份 sql文件;然后再执行:

以下为错误示范

mysql -u root -p xxx < xxxx20241012.sql

以上命令请认真仔细看好,每个字都需确认清楚;

毫无疑问,这句命令犯了一个巨大错误(备份还原不能对一个表操作)

正确命令为:

mysql -u root -p xxx2 < xxxx20241012.sql

由于错误的命令,我使用dump备份覆盖了线上数据库;而且由于还原耗时非常长;导致数据库丢失了2H的数据

本文介绍我的binlog数据还原过程;

解决方案

首先声明我的mysql版本是8025 ,服务所用框架为 spring boot 数据存储使用mybatis

很多教程都写的是用position来还原数据,这当然是标准操作,毕竟有mysqlbinlog命令原生支持

假如操作简单这么做当然省事效果好;但我其实是还原覆盖了数据库两次,服务器上生成了多个binlog文件;而且正式环境数据太多,我无法找到正确起止pos,我更担心这个操作会再次对数据库的数据产生不利影响;进而产生更糟糕的结果;

因此我决定退而求其次,放弃全表还原,仅从binlog中 找到重要数据 的SQL语句,仅还原重要数据.

虽然,数据库被之前的备份还原导致中间的数据丢失;但在还原的过程中;服务正常运行,数据还是曾经写入到了数据库中;

因此binlog会有正常流程产生的数据;因此只需找到此时的INSERT 和UPDATE语句即可修复数据问题.

数据处理过程中涉及非常多的步骤,由于我非专业dba,linux命令也不熟;多亏的chatgpt,很多问题 它能提供能运行的代码;为我节省了大量时间;强烈建议多用用.

操作步骤

找到异常时数据库所有SQL记录

• 首先,找到mysql 服务器数据存放路径;查看异常binlog文件,由于我多次还原库因此我这里会有多个binlog

ls /xxx/mysql -all

xxxx 14:00 binlog.000048

xxxx 15:00 binlog.000049

xxxx 16:00 binlog.000050

xxxx 17:00 binlog.000051

xxxx 18:00 binlog.000052

• 还原数据之前,可以先将binlog文件 下载到本地,然后用本地的mysqlbinlog命令解密及搜索处理;因为我这里数据量巨大而且完整解密出SQL语句之后的文件同样非常大;如此大的文件,操作建议本地在SSD上进行.

• 可以直接用less 查看 binlog;但是里面的SQL语句 base64加密了,因此使用以下命令解密

>mysqlbinlog --stop-datetime="2024-10-14 16:00:00" binlog.000050 --base64-output=decode-rows --verbose >coded_after50.sql

因为我这里数据多为方便追踪,每个binlog按照序号分别解密

...coded_after51.sql
...coded_after52.sql

使用less查看数据内容:

less code_after50.sql

#241012 16:29:35 server id 1  end_log_pos 485557996 CRC32 0x1bc4ffbf 	Table_map: `xxx`.`yyyy` mapped to number 212
# at 485557996
#241012 16:29:35 server id 1  end_log_pos 485558475 CRC32 0x7448527f 	Update_rows: table id 212 flags: STMT_END_F
### UPDATE `xxx`.`yyyy`
### WHERE
###   @1=zzz
###   @2='2024-10-12 16:29:25'
###   @3='xxxx'
###   @4=NULL
###   @5=NULL
###   @6=0
###   @7=NULL
###   @8='33'
###   @9=NULL

以上为我解密的sql文件示例;每个binlog文件都有大量的SQL语句,每个文件都大于2G;在windows中 即使使用logview也极难找到我想要的数据;

但linux 的grep却极其好用高效;因此我在powershell中输入bash 切换到wsl环境中来操作.

筛选出目标表的操作SQL语句

• 确认大致SQL

假如你直接执行grep 'yyyy' coded_after50.sql会发现大量不需要的数据,且没有上下文无法确定;此时你向ChatGPT求助;它会告诉你正确的命令;一番修改后就得到这样的语句

grep -A 50 yyyy decoded_after50.sql | grep -A 50 -B 25 "@2='2024-10-12"

此处解释一下:

-A 50表面返回表后面50行数据,方便查看是否是需要的SQL语句

| grep两次搜索 是从相关的表中找到出现问题时执行的SQL语句

-B 25返回二次搜索关键字前25行,方便后续处理

@2='2024-10-12'是我表中的创建时间,添加此条件后只返回当天的数据,如果想要过滤详细时间

​ 可以使用-E "@2='2024-10-12 15:[4-5]"使用正则查找15点40之后的数据

​ 使用-E "@2='2024-10-12 1[6-8]"查找16-18点的数据

多次调整过滤条件后添加 > 50.sql将结果写入到文件中 方便后续处理

所以现在的命令是这样的

grep -A 50 yyyy decoded_after50.sql | grep -A 50 -B 25 -E "@2='2024-10-12 15:[4-5]" >50.SQL

假如有多个文件想要合并到一起.可以使用>>操作符,如:

grep -A 50 yyyy decoded_after51.sql | grep -A 50 -B 25 -E "@2='2024-10-12 15:[4-5]" >>50.SQL

表示附加binlog51中的查找结果到50.SQL中

• 返回明确SQL语句

根据你的业务场景,你可以从50.SQL中找到大致的SQL语句,但是由于服务使用了mybatis导致有很多@符号,且文件中包括很多非SQL字符串,要转成所需SQL还需进一步处理

我这里给出我的简单处理的代码,使用C# 创建控制台即可使用

using Dapper;
using MySqlConnector;
using System.Data;
using System.Data.SqlClient;
using static System.Net.Mime.MediaTypeNames;
using System.Text.RegularExpressions;

namespace ConsoleApp1
{
    internal class Program
    {
        static void Main(string[] args)
        {
            // 连接字符串
            string connectionString =
                "server=xxxx";

            // 使用Dapper执行SQL语句
            using (IDbConnection connection = new MySqlConnection(connectionString))
            {               
                var users = connection.Query<User>("show columns from xxx;").Select(s => s.Field).ToList();

                string filePath; // Replace with your actual file path

                // 目标目录
                string directoryPath = @"G:\data\sql\";

                // 最大文件大小限制 (500 KB)
                long maxSizeInBytes = 500 * 1024;

                // 获取目录中所有文件
                string[] files = Directory.GetFiles(directoryPath);

                // 筛选大小小于500 KB 的文件
                var smallFiles = files.Where(file =>
                {
                    var info = new FileInfo(file);
                    return info.Length < maxSizeInBytes && info.Extension == ".sql";
                });
                foreach (var item in smallFiles)
                {
                    filePath = item;
                    // 输出文件路径
                    var fileINfo = new FileInfo(filePath);
                    var isInsert = false;
                    string outputFilePath = fileINfo.Name;

                    // 使用流读取文件
                    using (StreamReader reader = new StreamReader(filePath))
                    {
                        // 使用流写入新文件
                        using (StreamWriter writer = new StreamWriter(outputFilePath))
                        {
                            string line;
                            var listColumnName = new List<string>();

                            while ((line = reader.ReadLine()) != null)
                            {
                                // 查找 "###" 并替换后面的内容
                                int index = line.IndexOf("###");

                                var tableIndex = false;
                                var needColumns = false;
                                if (index >= 0)
                                {

                                    Match match = Regex.Match(line, @"`(\w+)`.`(\w+)`");

                                    if (match.Success)
                                    {
                                        listColumnName = connection.Query<User>($"show columns from {match.Groups[2].Value};").Select(s => s.Field).ToList();
                                        if (line.Contains("INSERT INTO"))
                                        {
                                            needColumns = true;
                                            isInsert=true;
                                        }
                                        else
                                        {
                                            isInsert = false;
                                        }
                                        tableIndex = true;
                                    }
                                    // 提取需要替换的部分
                                    string replacement = line.Substring(index + 3).Trim();
                                    string result;

                                    if (isInsert)
                                    {
                                        string pattern = @"@\d+="; // 正则表达式匹配 @数字=
                                        result = Regex.Replace(replacement, pattern, "");
                                    }
                                    else
                                    {

                                        // 正则表达式
                                        string pattern = @"@(\d+)";

                                        // 替换函数
                                        MatchEvaluator evaluator = match =>
                                        {
                                            // 获取匹配的数字
                                            int index2 = int.Parse(match.Groups[1].Value) - 1;

                                            // 检查数组索引是否有效
                                            if (index2 >= 0 && index2 < listColumnName.Count)
                                            {
                                                // 返回数组中对应元素的值
                                                return listColumnName[index2];
                                            }
                                            else
                                            {
                                                // 索引无效，返回原始字符串
                                                return match.Value;
                                            }
                                        };

                                        // 使用正则表达式替换
                                        result = Regex.Replace(replacement, pattern, evaluator);
                                    }


                                    // 替换内容
                                    line = line.Replace(line.Substring(index),
                                        result + (needColumns ? $"({string.Join(",", listColumnName)})" : ","));
                                }
                                else if (tableIndex)
                                {
                                    tableIndex = false;
                                    listColumnName = new List<string>();
                                    isInsert= false;
                                }

                                // 写入新文件
                                writer.WriteLine(line);
                            }
                        }
                    } 
                }
            }
        }

        public class User
        {
            public string Field { get; set; }
        }
    }
}

处理后在bin目录即可看到同名文件只需少量复制粘贴即可有完整SQL

这里,假如你想要标识哪些SQL没写入;可以使用grepwin搜索表名;然后根据行号逐个复制,防止遗漏

当然,你也可以写你自己的文本处理方法,只要调用show columns from xxx;找到表对应的字段 即可替换出正确的SQL

数据还原

按照以上的步骤找出所有表相关的SQL语句之后,我从正式库同步一份目标表结构的空库到本地;

然后依时间执行SQL语句,即可得到最终结果的表;

中间因为大文本搜索,可能会有重复数据,本地执行SQL后数据库会帮我们检查数据的正确性,

根据逻辑,检查数据缺失或者关联的表数据

全部流程完成之后;

在本地的新测试库即可得到完整的修复数据拷贝,此时将所有数据导出成SQL语句

再补充上事务代码:

START TRANSACTION;
...INSERT
COMMIT;

多次检查之后,就能上线修复了.

参考资料

• https://kousikraj.me/2014/07/19/how-to-recover-data-table-level-using-mysql-bin-log-file-mysqlbinlog/
• Mysql 通过 binlog日志 恢复数据（数据搞丢看过来）
• 带你了解 MySQL Binlog 不为人知的秘密
• ChatGPT