TiDB 利用binlog 恢复-反解析binlog

2024-04-30 07:44
文章标签 tidb 恢复 解析 binlog

本文主要是介绍TiDB 利用binlog 恢复-反解析binlog,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

我们知道TiDB的binlog记录了所有已经执行成功的dml语句,类似mysql binlog row模式

,TiDB官方也提供了reparo可以进行解析binlog,如下所示:

[2024/04/26 20:58:02.136 +08:00] [INFO] [config.go:153] ["Parsed start TSO"] [ts=449217508147200000]
[2024/04/26 20:58:02.136 +08:00] [INFO] [config.go:160] ["Parsed stop TSO"] [ts=449222855884800000]
schema: coupon_trade; table: coupon_trade_record; type: Insert
coupon_id(varchar): 1
merchant_id(varchar): 2
coupon_code(varchar): 4
customer_id(varchar): 4
customer_mobile_no(varchar): 5
trade_channel(varchar): 03
trade_store_id(varchar): 1440040
trade_store_name(varchar): <nil>
bill_type(tinyint): 0
bill_code(varchar): 6
external_order_id(varchar): <nil>
receive_channel_code(varchar): 105
coupon_card_type(tinyint): 0
coupon_type(tinyint): 1
receive_type(smallint): 6
receive_src_code(varchar): <nil>
process_type(tinyint): 2
order_food_type(varchar): 1
order_mode(varchar): 0
is_give_away(tinyint): 0
is_available(tinyint): 1
is_deleted(bigint): 0
updated_date(datetime): 2024-04-21 00:00:00
created_date(datetime): 2024-04-21 00:00:00
updated_user(varchar): default
created_user(varchar): default
version(int): 1
schema: coupon_trade; table: coupon_trade_record; type: Insert
coupon_id(varchar): 2
merchant_id(varchar): <nil>
coupon_code(varchar): 5
customer_id(varchar): 6
customer_mobile_no(varchar): 7
trade_channel(varchar): 03
trade_store_id(varchar): 1420452
trade_store_name(varchar): <nil>
bill_type(tinyint): 0
bill_code(varchar): 8
external_order_id(varchar): <nil>
receive_channel_code(varchar): 03
coupon_card_type(tinyint): 0
coupon_type(tinyint): 0
receive_type(smallint): 4
receive_src_code(varchar): <nil>
process_type(tinyint): 2
order_food_type(varchar): 1
order_mode(varchar): 0
is_give_away(tinyint): 0
is_available(tinyint): 1
is_deleted(bigint): 0
updated_date(datetime): 2024-04-21 00:00:00
created_date(datetime): 2024-04-21 00:00:00
updated_user(varchar): default
created_user(varchar): default
version(int): 1

另外reparo 支持print 和mysql 两种模式,print只做解析打印到标准输出,不执行 SQL,mysql:是直接再下游数据库执行SQL

但是我们有的时候需要进去反解析,比如我们误删除了一些数据,我们要把误删除的数据解析成INSERT语句,这怎么办呢,TIDB目前不提供这种反解析工具,于是自己写了一个工具进行解析,代码如下:

package mainimport ("bufio""encoding/json""flag""fmt""log""os""regexp""strings""time"
)var (binlogFile stringschema     stringtable      stringsqlType    stringwhereSql   stringlogPath    stringfiledRe    = regexp.MustCompile(`\(.*?\)+:.*`)logOutFile *log.Logger
)type filed struct {Name  string `json:"name"`Value string `json:"value"`
}
type Parser struct {lines []filed
}//	type lists struct {
//		filed []filed
//	}
func compressStr(str string) string {str = strings.ReplaceAll(str, " ", "")r := strings.NewReplacer("\r", "", "\n", "")str = r.Replace(str)//匹配一个或多个空白符的正则表达式reg := regexp.MustCompile("\\s+")return reg.ReplaceAllString(str, "")
}func main() {flag.StringVar(&binlogFile, "binlogFile", "", "binlog日志文件路径")flag.StringVar(&schema, "schema", "", "要解析的数据库名称")flag.StringVar(&table, "table", "", "要解析的表名称")flag.StringVar(&sqlType, "sqlType", "", "要解析的dml类型")flag.StringVar(&logPath, "logPath", "", "输出文件路径名称")flag.Parse()if binlogFile == "" {log.Println("请输入binlog日志文件路径...")return}if schema == "" {log.Println("请输入要解析的数据库名称...")return}if table == "" {log.Println("请输入要解析的表名称...")return}if sqlType == "" {log.Println("请输入要解析的dml类型...")return}if logPath == "" {log.Println("请输入输出日志文件路径...")return}outSlowLogFile(logPath)file, err := os.Open(binlogFile)if err != nil {log.Println("读取文件失败...")return}schemaRe := fmt.Sprintf("schema:%v;table:%v;type:%v", schema, table, sqlType)//schemaRe2 := fmt.Sprintf("schema:%v;table:%v;type:%v", schema, table, "Insert")defer file.Close()scanner := bufio.NewScanner(file)scanner.Buffer(make([]byte, 1024*1024), 1024*1024*10)inHeader := falsevar f filedvar array []stringvar time2, _ = time.Parse("2006-01-02 15:04:05", "2024-04-01 00:00:00")fields := ""for scanner.Scan() {line := scanner.Text()if compressStr(line) == compressStr(schemaRe) {if fields != "" {fields = fmt.Sprintf("{%v}", fields)array = append(array, fields)}fields = ""inHeader = truecontinue} else if strings.Contains(line, "sync binlog success") || strings.Contains(line, "read file end") {inHeader = falsecontinue} else {if inHeader {if filedRe.MatchString(line) {f.Name = strings.Split(line, "(")[0]tmpValue := strings.Split(line, ": ")if len(tmpValue) == 1 {inHeader = falsecontinue} else {f.Value = strings.Split(line, ": ")[1]}if fields == "" {fields = fmt.Sprintf("\"%s\":\"%s\" ", f.Name, f.Value)} else {fields += fmt.Sprintf(",\"%s\":\"%s\"", f.Name, f.Value)}} else {inHeader = falsecontinue}}}}if fields != "" {fields = fmt.Sprintf("{%v}", fields)array = append(array, fields)}if err := scanner.Err(); err != nil {log.Println(err)}for i := 0; i < len(array); i++ {//decoder := json.NewDecoder(strings.NewReader(array[i]))////for key, value := range JsonToMap(array[i]) {//	fmt.Printf("键:%v,值:%d\n", key, value)//}m := make(map[string]string)err = json.Unmarshal([]byte(array[i]), &m)if err != nil {fmt.Printf("Unmarshal with error: %+v", err)}keys := ""values := ""isExec := false//newKeys := make([]string, 0, len(m))//for k := range m {//	newKeys = append(newKeys, k)//}对切片进行排序//sort.Strings(newKeys)for key, value := range m {if keys == "" {keys = fmt.Sprintf("%v", key)if value == "<nil>" {values = fmt.Sprintf("%v", "NULL")} else {values = fmt.Sprintf("'%v'", value)}} else {keys += fmt.Sprintf(",%v", key)if value == "<nil>" {values += fmt.Sprintf(",%v", "NULL")} else {values += fmt.Sprintf(",'%v'", value)}}if key == "updated_date" {isExec = trueupdateTime, _ := time.Parse("2006-01-02 15:04:05", value)if updateTime.After(time2) {isExec = true} else {isExec = false}}}if isExec {inSql := fmt.Sprintf("insert into coupon_trade_record(%v) values(%v);", keys, values)logOutFile.Println(inSql)}}
}func outSlowLogFile(outFile string) {outFilePath, err := os.OpenFile(outFile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0766)if err != nil {log.Println(fmt.Sprintf("创建输出文件失败:%s", err))return}logOutFile = log.New(outFilePath, "", log.Lmsgprefix)
}

通过代码可以将DELETE sql直接转成insert 语句:

 至此完成将数据重新插入到业务库里面,即可完成恢复

这篇关于TiDB 利用binlog 恢复-反解析binlog的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/948279

相关文章

Springboot @Autowired和@Resource的区别解析

《Springboot@Autowired和@Resource的区别解析》@Resource是JDK提供的注解,只是Spring在实现上提供了这个注解的功能支持,本文给大家介绍Springboot@... 目录【一】定义【1】@Autowired【2】@Resource【二】区别【1】包含的属性不同【2】@

SpringCloud动态配置注解@RefreshScope与@Component的深度解析

《SpringCloud动态配置注解@RefreshScope与@Component的深度解析》在现代微服务架构中,动态配置管理是一个关键需求,本文将为大家介绍SpringCloud中相关的注解@Re... 目录引言1. @RefreshScope 的作用与原理1.1 什么是 @RefreshScope1.

Java并发编程必备之Synchronized关键字深入解析

《Java并发编程必备之Synchronized关键字深入解析》本文我们深入探索了Java中的Synchronized关键字,包括其互斥性和可重入性的特性,文章详细介绍了Synchronized的三种... 目录一、前言二、Synchronized关键字2.1 Synchronized的特性1. 互斥2.

Java的IO模型、Netty原理解析

《Java的IO模型、Netty原理解析》Java的I/O是以流的方式进行数据输入输出的,Java的类库涉及很多领域的IO内容:标准的输入输出,文件的操作、网络上的数据传输流、字符串流、对象流等,这篇... 目录1.什么是IO2.同步与异步、阻塞与非阻塞3.三种IO模型BIO(blocking I/O)NI

Python 中的异步与同步深度解析(实践记录)

《Python中的异步与同步深度解析(实践记录)》在Python编程世界里,异步和同步的概念是理解程序执行流程和性能优化的关键,这篇文章将带你深入了解它们的差异,以及阻塞和非阻塞的特性,同时通过实际... 目录python中的异步与同步:深度解析与实践异步与同步的定义异步同步阻塞与非阻塞的概念阻塞非阻塞同步

使用Python实现网络设备配置备份与恢复

《使用Python实现网络设备配置备份与恢复》网络设备配置备份与恢复在网络安全管理中起着至关重要的作用,本文为大家介绍了如何通过Python实现网络设备配置备份与恢复,需要的可以参考下... 目录一、网络设备配置备份与恢复的概念与重要性二、网络设备配置备份与恢复的分类三、python网络设备配置备份与恢复实

Redis中高并发读写性能的深度解析与优化

《Redis中高并发读写性能的深度解析与优化》Redis作为一款高性能的内存数据库,广泛应用于缓存、消息队列、实时统计等场景,本文将深入探讨Redis的读写并发能力,感兴趣的小伙伴可以了解下... 目录引言一、Redis 并发能力概述1.1 Redis 的读写性能1.2 影响 Redis 并发能力的因素二、

Spring MVC使用视图解析的问题解读

《SpringMVC使用视图解析的问题解读》:本文主要介绍SpringMVC使用视图解析的问题解读,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录Spring MVC使用视图解析1. 会使用视图解析的情况2. 不会使用视图解析的情况总结Spring MVC使用视图

MySQL使用binlog2sql工具实现在线恢复数据功能

《MySQL使用binlog2sql工具实现在线恢复数据功能》binlog2sql是大众点评开源的一款用于解析MySQLbinlog的工具,根据不同选项,可以得到原始SQL、回滚SQL等,下面我们就来... 目录背景目标步骤准备工作恢复数据结果验证结论背景生产数据库执行 SQL 脚本,一般会经过正规的审批

利用Python和C++解析gltf文件的示例详解

《利用Python和C++解析gltf文件的示例详解》gltf,全称是GLTransmissionFormat,是一种开放的3D文件格式,Python和C++是两个非常强大的工具,下面我们就来看看如何... 目录什么是gltf文件选择语言的原因安装必要的库解析gltf文件的步骤1. 读取gltf文件2. 提