准备工作
1.操作的机器需要安装python3.6或其他版本
2.操作机器需要安装mysql8.0，mysql/bin目录下有mysqlbinlog工具文件
3.把我写的python脚本放到服务器中
开始干活
b.python 内容
#_*_ coding:utf-8 _*_
# 作者yichen qq2782268022
import sys
import os
import io
import re
binlogfile = sys.argv[1]
database_name = sys.argv[2]
table_name = sys.argv[3]
print('--database="'+database_name+'"')
# binlogfile="b472in.txt"
rootname="/www/server/mysql/bin/mysqlbinlog"
def format_binlog():
os.system(rootname+' --start-datetime="2021-11-25 16:48:00" --skip-gtids --stop-datetime="2021-11-25 19:50:00" --database='+database_name+' -v -v --base64-output=DECODE-ROWS /www/server/mysql/bin/'+binlogfile+' > '+binlogfile+'.log')
#需要处理的语句
# confMethod=["INSERT","UPDATE","DELETE","ALTER","CREATE","REPLACE"]
confMethod=["INSERT","UPDATE"]
def isMethod(method,row):
if method=="INSERT" and "### INSERT" == row[0:10]:
return True
elif method =="UPDATE" and "### UPDATE" == row[0:10]:
return True
elif method=="DELETE" and "### DELETE" == row[0:10]:
return True
elif method =="ALTER" and "ALTER" == row[0:5]:
return True
elif method =="CREATE" and "CREATE" == row[0:5]:
return True
elif method =="REPLACE" and "REPLACE"== row[0:7]:
return True
else:
return False
def getMethod(methodArr,row):
for method in methodArr:
if isMethod(method, row):
return method
return None
def pickupbinlog():
f = io.open(binlogfile+'.log','r')
fw = io.open(binlogfile+'_.log','a')
# 记录上一行首3个字符
priv_str = ''
#段落分割线首行
priv_line = ''
goal_flag = 0
# 测试到某行停止执行
nub=0
# where条件部分
whereRow=""
# # set部分
# setRow=""
# 部分标识
stateRow="" #WHERE
# stateAll="UPDATE DELETE INSERT ALTER"
method="" #UPDATE
# # 首部分
# headRow=""
# # 其他部分
# otherRow=""
rowData=""
lastState=0
nub=1
for row in f:
lineMethod=getMethod(confMethod,row)
# if(method=="UPDATE"):
### create|### replace|### insert|### alter|### update
# # 处理首行 对批量input不准确
linRow=row
if lineMethod!=None:
# 上条语句如果没结束，添加;字符作为结束符
if(len(method)>1):
if(rowData.find(";")==-1):
if method =="INSERT":
fw.write(");\n")
elif method == "UPDATE" or method=="DELETE":
fw.write(" WHERE "+whereRow+"\n")
else:
fw.write(";\n")
# if priv_str !="###":
# fw.write(priv_line)
method=""
stateRow=""
if database_name in row and table_name in row:
# if priv_str !="###":
# fw.write(priv_line)
if(row.find("`"+database_name+"`.")):
row=row.replace("`"+database_name+"`.","")
goal_flag=1
if lineMethod=="INSERT":
method="INSERT"
rowData=row.replace("### INSERT INTO","INSERT IGNORE INTO")+" VALUES("
# print(rowData)
# if(headRow.find("`"+database_name+"`.")):
# # headRow=headRow.translate(str.maketrans('','',"`"+database_name+"`."))
# headRow=headRow.replace("`"+database_name+"`.","")
elif lineMethod=="UPDATE":
method="UPDATE"
rowData=row.replace("### UPDATE","UPDATE IGNORE")
# print(row)
elif lineMethod=="DELETE":
method="DELETE"
# print(row)
rowData=row.replace("### DELETE","DELETE")
elif lineMethod=="ALTER":
method="ALTER"
rowData = row
elif lineMethod =="CREATE":
method="CREATE"
rowData=row.replace("### CREATE","CREATE")
elif lineMethod=="REPLACE":
method="REPLACE"
rowData = row
else:
rowData = row
rowData=rowData.replace("### ","")
else:
method=""
rowData="";
stateRow=""
goal_flag=0
elif row[0:3] != '###' and priv_str == '###':
stateRow=""
# 上条语句如果没结束，添加;字符作为结束符
if(len(method)>1):
if(rowData.find(";")==-1):
if method =="INSERT":
fw.write(");\n")
elif method == "UPDATE" or method=="DELETE":
fw.write(" WHERE "+whereRow+"\n")
else:
fw.write(";\n")
fw.write(row);
rowData=""
method="";
goal_flag=0
# if database_name in row and table_name in row:
# # if row[0:3] == '###' and database_name in row and table_name in row:
# goal_flag = 1
# 处理目标操作
elif row[0:3] == '###' and goal_flag==1:
# 删除最后吃出现的注释/*之后部分
if row.rfind( '/*' )>0:
row=row[ 0 : row.rfind( '/*' ) ]
# 记录本行 下次首行进行判断
if( "### SET"==row[0:7]):
stateRow="SET"
if(method!="INSERT"):
rowData=row[4:len(row)]
# rowData=row[4:len(row)]
elif("### WHERE" == row[0:9]):
stateRow="WHERE"
rowData=""
else:
rowlen=len(row)
# print ("处理目标操作")
#这是我的特殊需求去除结尾空格
# row=row.rstrip()
if("WHERE" == stateRow):
rowData="";
if("@1=" in row):
whereRow=row[4:rowlen]+";"
elif("SET" == stateRow):
# print(row)
if("@" in row):
if method=="INSERT":
end=row.find( '=' )
if end>0:
# 对于一个节点@1 判断
if row.find("@1=")>-1 and row.find("@1=")<10:
rowData=row[ end+1: len(row) ]
else:
rowData=","+row[ end+1: len(row) ]
# rowData="("+setRow+",";
# fw.write(rowData);
# setRow+row+","
elif method=="UPDATE":
# 对于一个节点下匹配input操作，根据重复@1 判断
if row.find("@1=")>-1 and row.find("@1=")<10:
rowData=row[4:rowlen]
else:
rowData=","+row[4:rowlen]
else:
rowData=row[4:rowlen]
else:
rowData=row[4:rowlen]
# print(row)
elif goal_flag==1:
if "SET @@SESSION.GTID_NEXT" in row or "BEGIN" ==row[0:5] or "SET TIMESTAMP" == row[0:13]:
rowData=""
else:
rowData=row
if(len(rowData)>1):
fw.write(rowData)
rowData=""
priv_line = linRow
priv_str = priv_line[0:3]
# elif(len(row)>2):
# priv_str = priv_line[0:3]
# fw.write(row)
# if nub>7:
# # print(setRow)
# break;
f.close()
fw.close()
if __name__ == '__main__':
# python2.7 pickupbinlog.py mysql-bin.001051 dbname tablename
# python3 pickupbinlog.py mysql-bin.001051 dbname tablename
format_binlog()
pickupbinlog()
在b.python 内容里需要修改下/www/server/mysql/bin/mysqlbinlog 的参数。根据你的需求参考mysqlbinlog官方文档配置。我的案例是按启止时间和数据库名去查找的
然后 执行命令
python b.py mysqlbinlog日志文件名 数据库名 数据库的表名
#python.py 可以配置想要处理的方法参数， confMethod=["INSERT","UPDATE","DELETE","ALTER","CREATE","REPLACE"]
python b.py mysql-bin.000473 dbname tablename
查看关键字 你要的区域，获得所在行号，记录下来
cat mysql-bin.000473_.log -n |grep -A 50 "ALTER" | head -n 200
截取第一个行到行号的内容
sed -n '1,110095p' mysql-bin.000472_.log>b472a.log
截取行号 到结尾的内容
sed -n "110095,&p" mysql-bin.000472_.log>b472b.log
把结果内容的@字段替换成数据库对应的字段
用sql语句执行
#假设数据库名：beiyao-zt-test
#假设数据表名：'zt_product_base'
set @irowa =0 ;
set @irowb =51 ;
SELECT GROUP_CONCAT(CONCAT("s/@",@irowa:=@irowa+1,'=/',COLUMN_NAME,'=/g') SEPARATOR ";") as a FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = 'beiyao-zt-test' AND TABLE_NAME = 'zt_product_base'
得到如下@1--对应字段的字符串
s/@1=/id=/g;s/@2=/brand_name=/g;
linux命令 替换生成数据库可执行的sql文件
sed 's/@1=/id=/g;s/@2=/upc=/g;' b472a.log>b472a.sql
然后导入sql文件就可以啦
备用一个php文件
<?php
ini_set("error_reporting","E_ALL & ~E_NOTICE");
$sql_file="mysql-bin.000472"; //完整的sql文件
$tab_name="`beiyaozhongtai`.`zt_product_base`"; //要提取的表
$sql = new SplFileObject($sql_file);
$log=false;
foreach($sql as $line){ //逐行读取
if ( preg_match("/^(#211125 )/", $line) ) { //当前行找到匹配
$timestamp=$line;
}
if ( preg_match("/(replace|alter)([A-Z_\-\.\s\`]+){$tab_name}/i", $line) ) { //当前行找到匹配
// if ( preg_match("/(create|delete|replace|insert|alter|update)([A-Z_\-\.\s\`]+){$tab_name}/i", $line) ) { //当前行找到匹配
// if ( preg_match("/(### create|replace|### insert|alter|### update)([A-Z_\-\.\s\`]+){$tab_name}/i", $line) ) { //当前行找到匹配
// print_r($timestamp."\r\n");
file_put_contents($sql_file."_out.sql", $timestamp, FILE_APPEND); //时间戳
$log=true;
}
if ($log) {
file_put_contents($sql_file."_out.sql", $line, FILE_APPEND);
}
if ( $log && preg_match("/\/\*\!\*\/;/",$line) ){
$log=false;
}
}
echo "导出完毕！\n";
几个参数
if ( preg_match("/^(#211125 )/", $line) ) { //当前行找到匹配
#211125 这个参数需要根据实际binlog文件修改
执行命令
php b.php
网友评论