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
2
...coded_after51.sql
...coded_after52.sql

使用less查看数据内容:

less code_after50.sql

1
2
3
4
5
6
7
8
9
10
11
12
13
14
#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# 创建控制台即可使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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搜索表名;然后根据行号逐个复制,防止遗漏

image-20241014012955156

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

数据还原

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

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

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

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

全部流程完成之后;

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

再补充上事务代码:

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

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

参考资料