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加密了,因此使用以下命令解密

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

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

1
...coded_after51.sql
2
...coded_after52.sql

使用less查看数据内容:

less code_after50.sql

1
#241012 16:29:35 server id 1  end_log_pos 485557996 CRC32 0x1bc4ffbf   Table_map: `xxx`.`yyyy` mapped to number 212
2
# at 485557996
3
#241012 16:29:35 server id 1  end_log_pos 485558475 CRC32 0x7448527f   Update_rows: table id 212 flags: STMT_END_F
4
### UPDATE `xxx`.`yyyy`
5
### WHERE
6
###   @1=zzz
7
###   @2='2024-10-12 16:29:25'
8
###   @3='xxxx'
9
###   @4=NULL
10
###   @5=NULL
11
###   @6=0
12
###   @7=NULL
13
###   @8='33'
14
###   @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# 创建控制台即可使用

1
using Dapper;
2
using MySqlConnector;
3
using System.Data;
4
using System.Data.SqlClient;
5
using static System.Net.Mime.MediaTypeNames;
6
using System.Text.RegularExpressions;
7

8
namespace ConsoleApp1
9
{
10
    internal class Program
11
    {
12
        static void Main(string[] args)
13
        {
14
            // 连接字符串
15
            string connectionString =
16
                "server=xxxx";
17

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

23
                string filePath; // Replace with your actual file path
24

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

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

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

34
                // 筛选大小小于500 KB 的文件
35
                var smallFiles = files.Where(file =>
36
                {
37
                    var info = new FileInfo(file);
38
                    return info.Length < maxSizeInBytes && info.Extension == ".sql";
39
                });
40
                foreach (var item in smallFiles)
41
                {
42
                    filePath = item;
43
                    // 输出文件路径
44
                    var fileINfo = new FileInfo(filePath);
45
                    var isInsert = false;
46
                    string outputFilePath = fileINfo.Name;
47

48
                    // 使用流读取文件
49
                    using (StreamReader reader = new StreamReader(filePath))
50
                    {
51
                        // 使用流写入新文件
52
                        using (StreamWriter writer = new StreamWriter(outputFilePath))
53
                        {
54
                            string line;
55
                            var listColumnName = new List<string>();
56

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

62
                                var tableIndex = false;
63
                                var needColumns = false;
64
                                if (index >= 0)
65
                                {
66

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

69
                                    if (match.Success)
70
                                    {
71
                                        listColumnName = connection.Query<User>($"show columns from {match.Groups[2].Value};").Select(s => s.Field).ToList();
72
                                        if (line.Contains("INSERT INTO"))
73
                                        {
74
                                            needColumns = true;
75
                                            isInsert=true;
76
                                        }
77
                                        else
78
                                        {
79
                                            isInsert = false;
80
                                        }
81
                                        tableIndex = true;
82
                                    }
83
                                    // 提取需要替换的部分
84
                                    string replacement = line.Substring(index + 3).Trim();
85
                                    string result;
86

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

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

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

104
                                            // 检查数组索引是否有效
105
                                            if (index2 >= 0 && index2 < listColumnName.Count)
106
                                            {
107
                                                // 返回数组中对应元素的值
108
                                                return listColumnName[index2];
109
                                            }
110
                                            else
111
                                            {
112
                                                // 索引无效，返回原始字符串
113
                                                return match.Value;
114
                                            }
115
                                        };
116

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

121

122
                                    // 替换内容
123
                                    line = line.Replace(line.Substring(index),
124
                                        result + (needColumns ? $"({string.Join(",", listColumnName)})" : ","));
125
                                }
126
                                else if (tableIndex)
127
                                {
128
                                    tableIndex = false;
129
                                    listColumnName = new List<string>();
130
                                    isInsert= false;
131
                                }
132

133
                                // 写入新文件
134
                                writer.WriteLine(line);
135
                            }
136
                        }
137
                    }
138
                }
139
            }
140
        }
141

142
        public class User
143
        {
144
            public string Field { get; set; }
145
        }
146
    }
147
}

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

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

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

数据还原#

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

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

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

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

全部流程完成之后;

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

再补充上事务代码:

1
START TRANSACTION;
2
...INSERT
3
COMMIT;

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

引言#

解决方案#

操作步骤#

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

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

数据还原#

参考资料#