MySQL数据归档小工具推荐及优化--mysql_archiver

一.主要概述

MySQL数据库归档历史数据主要可以分为三种方式：一.创建编写SP、设置Event；二.通过dump导入导出；三.通过pt-archiver工具举行归档。第一种方式每每受限于同实例要求，每每被各人舍弃。第二种，性能相对较好，但是归档表较多时运维也是比力头疼的事。所以许多DBA每每接纳第三种方式--pt-archiver。
pt-archiver是Percona-Toolkit工具集中的一个组件，是一个主要用于对MySQL表数据举行归档和清除的工具。它可以将数据归档到另一张表大概是一个文件中。pt-archiver在清除表数据的过程中并不会影响OLTP事务的查询性能。对于数据的归档，它可以归档到另一台服务器上的另一张表，也可归档到一个文件中。
有一个网友通过Python开辟了将pt-archiver打包成了一个小工具—mysql_archiver，它基本实现了归档作业的设置化部署，使用起来挺不错。"MySQL_archiver基本上实现了数据归档的自动运转，统一的归档任务调理管理、自动监控和预警、自动生成报表。在一定程度上节省了生产力，提高了运维效率。"
github地址：GitHub - dbarun/mysql_archiver: MySQL数据归档小工具

二.pt-archiver主要参数

参数	默认值	参数表明
--check-slave-lag		指定主从复制延迟大于选项'--max-lag'指定的值之后暂停归档利用。默认情况下，工具会检查全部的从库，但该选项只作用于指定的从库(通过DSN连接方式)。
--check-interval	1s	假如同时指定了选项'--check-slave-lag'，则该选项指定的时间为工具发现主从复制延迟时暂停的时间。每举行利用100行时举行一次检查。
--[no]check-charset	yes	指定检查确保数据库连接时字符集和表字符集相同。
--commit-each		指定按每次获取和归档的行数举行提交，该选项会禁用选项'--txn-size'。 在每次获取表数据并举行归档之后，在获取下一次数据和选项'--sleep'指定的休眠时间之前，举行事务提交和刷新选项'--file'指定的文件，通过选项'--limit'控制事务的大小。
--host，-h		指定连接的数据库IP地址。
--port，-P		指定连接的数据库Port端口。
--user，-u		指定连接的数据库用户。
--password，-p		指定连接的数据库用户密码。
--socket，-S		指定使用SOCKET文件连接。
--databases，-d		指定连接的数据库
--source		指定需要举行归档利用的表，该选项是必须指定的选项，使用DSN方式表示。
--dest		指定要归档到的目的端表，使用DSN方式表示。 假如该选项没有指定的话，则默认与选项'--source'指定源端表为相同表。
--where		指定通过WHERE条件语句指定需要归档的数据，该选项是必须指定的选项。不需要加上'WHERE'关键字，假如确实不需要WHERE条件举行限制，则指定'--where 1=1'。
--file		指定表数据需要归档到的文件。使用类似MySQL DATE_FORMAT()格式化命名方式。 文件内容与MySQL中SELECT INTO OUTFILE语句使用相同的格式，文件命名选项如下所示： ' %Y：年，4位数(Year, numeric, four digits) %m：月，2位数(Month, numeric (01..12)) %d：日，2位数(Day of the month, numeric (01..31)) %H：小时(Hour (00..23)) %i：分钟(Minutes, numeric (00..59)) %s：秒(Seconds (00..59)) %D：数据库名(Database name) %t：表名(Table name) 例如：--file '/var/log/archive/%Y-%m-%d-%D.%t' '
--output-format		指定选项'--file'文件内容输出的格式。 默认不指定该选项是以制表符举行字段的分隔符，假如指定该选项，则使用','(逗号)作为字段分隔符，使用'"'(双引号)将字段括起。用法示例：'--output-format=dump'。
--limit	1	指定每条语句获取表和归档表的行数。
--max-lag	1s	指定答应主从复制延迟时长的最大值，单位秒。假如在每次获取行数据之后主从延迟高出指定的值，则归档利用将暂停执行，暂停休眠时间为选项'--check-interval'指定的值。待休眠时间结束之后再次检查主从延迟时长，检查方法是通过从库查询的'Seconds_Behind_Master'值来确定。假如主从复制延迟一直大于该参数指定值大概从库制止复制，则利用将一直等候直到从库重新启动并且延迟小于该参数指定值.
--no-delete		指定不删除已被归档的表数据。
--progress		指定每多少行打印进度信息，打印当前时间，已用时间以及多少行举行归档。
--purge		指定执行的清除利用而不是归档利用。答应忽略选项'--dest'和'--file'举行利用，假如只是清除利用可以结合选项'--primary-key-only'会更高效。
--replace		指定写入选项'--dest'指定目的端表时改写INSERT语句为REPLACE语句。
--retries	1	指定归档利用碰到死锁或超时的重试次数。当重试次数高出该选项指定的值时，工具将报错退出。
--run-time		指定工具归档利用在退出之前需要运行的时间。答应的时间后缀名为s=秒，m=分，h=小时，d=天，假如没指定，默以为s。
--[no]safe-auto-increment	yes	指定不使用自增列(AUTO_INCREMENT)最大值对应的行举行归档。 该选项在举行归档清除时会额外添加一条WHERE子句以防止工具删除单列升序字段具有的具有AUTO_INCREMENT属性最大值的数据行，为了在数据库重启之后还能使用到AUTO_INCREMENT对应的值，但这会引起无法归档或清除字段对应最大值的行。
--sleep		指定工具在通过SELECT语句获取归档数据需要休眠的时间，默认值是不举行休眠。在休眠之前事务并不会提交，并且选项'--file'指定的文件不会被刷新。假如指定选项'--commit-each'，则在休眠之前会举行事务提交和文件刷新。
--statistics		指定工具网络并打印利用的时间统计信息。
--txn-size	1	指定每个事务处置惩罚的行数。假如是0则禁用事务功能。
--why-quit		指定工具打印当非因完成归档行数退出的原因。 在执行一个自动归档任务时该选项与选项'--run-time'一起使用非常方便，这样可以确定归档任务是否在指定的时间内完成。假如同时指定了选项'--statistics'，则会打印全部退出的原因。
--skip-foreign-key-checks		指定使用语句SET FOREIGN_KEY_CHECKS = 0禁用外键检查。

三,安装部署mysql_archiver常见错误

1. 执行python db_archive_exec.py 127.0.0.1 db123 报错

1. Traceback (most recent call last):
2.   File "/data/XXXXXX/mysql_archiver-master/db_archive_exec.py", line 7, in <module>
3.     import db_conn
4.   File "/data/XXXXXX/mysql_archiver-master/db_conn.py", line 4, in <module> import MySQLdb ImportError: No module named MySQLdb

复制代码

 办理方案:
Step 1 底子情况

1. rm -rf /etc/yum.repos.d/epel*
2. yum install postgresql-devel python3-devel postgresql-devel python-devel

复制代码

Step 2 安装setuptools

1. （1）    下载    setuptools-39.2.0.zip
2. （2）    解压安装 python setup.py build
3.                 python setup.py install

复制代码

step 3 安装pip

1. （1）    下载 pip-1.5.4.tar.gz
2. （2）    安装 python setup.py install

复制代码

step 4 安装MySQL-python

1. yum install MySQL-python

复制代码

2.报错缺少pt-archiver工具

1. sh: pt-archiver: command not found

复制代码

办理方案：
下载percona-toolkit-3.2.1_x86_64.tar.gz,安装，并且设置个软连接。
例如:

1. ln -s /data/XXXXXXXX/percona-toolkit-3.2.1/bin/pt-archiver /usr/local/bin/pt-archiver

复制代码

3.缺少文档

1. <type 'exceptions.Exception'>
2. [Errno 2] No such file or directory: u'/software/python_script/db_archive_XXXXdb_XXXtablename.log'

复制代码

办理方案:
创建文档: /software/python_script

4.字符集问题

1. Character set mismatch: --source DSN uses utf8, table uses utf8mb4.  You can disable this check by specifying --no-check-charset.

复制代码

办理方案:
修改db_archive_exec.py可执行文件，针对pt-archiver添加参数 --no-check-charset

 5.参数设置报错

1. Usage: pt-archiver [OPTIONS] --source DSN --where WHERE
3. Errors in command-line arguments:
4.   * --txn-size and --commit-each are mutually exclusive.
6. pt-archiver nibbles records from a MySQL table.  The --source and --dest
7. arguments use DSN syntax; if COPY is yes, --dest defaults to the key's value
8. from --source.  For more details, please use the --help option, or try 'perldoc
9. /usr/local/bin/pt-archiver' for complete documentation.

复制代码

 办理方案:
pt-archiver去掉参数 --commit-each

 6.Cannot find encoding "utf8mb4"
报错信息:

1. Cannot find encoding "utf8mb4" at /usr/local/bin/pt-archiver line 6711.

复制代码

 办理方案:
pt-archiver去掉参数 --bulk-insert --bulk-delete

 7.执行 db_archive_exec.py 报错
报错信息

1. <type 'exceptions.Exception'>
2. 'ascii' codec can't decode byte 0xe6 in position 0: ordinal not in range(128)

复制代码

办理方案
修改优化的代码部分，不要有中文，即剔除添加的中文，修改为英文即可。

8.执行db_archive_monitor.py报错

1. Traceback (most recent call last):
2.   File "/data/XXXXX/mysql_archiver-master/db_archive_monitor.py", line 5, in <module>
3.     from db_log import db_archive_monitor_log
4. ImportError: No module named db_log

复制代码

办理方案
转到目录db_log,执行如下脚本

1. echo '' > __init__.py

复制代码

生成__init__.py文件
9. crontab执行python脚本问题
调试和手动执行，都没有问题，但是发现没有成功执行备份。
为了定位问题，我们可以将cron执行情况，打印初始处置惩罚。将crontab调解如下:
调解前:

1. 34 11 * * * python /data/mysql_archiver/mysql_archiver-master/db_archive_exec.py 122.122.122.122 qqorderdb

复制代码

调解后

1. 34 11 * * * python /data/mysql_archiver/mysql_archiver-master/db_archive_exec.py 122.122.122.122 qqorder >> /data/mysql_archiver/mysql_archiver-master/db_log/logs/crontab_archiver.log 2>&1

复制代码

查看执行log文件，报错信息如下：

1. [laobao@qqorder]# tail -100 crontab_archiver.log
2. sh: pt-archiver: command not found
3. sh: pt-archiver: command not found
4. sh: pt-archiver: command not found
5. sh: pt-archiver: command not found
6. sh: pt-archiver: command not found

复制代码

问题明白了，办理方案如下:
调解前:

调解后:

10.执行 db_archive_exec.py 大概碰到其他的报错
A.报错如下

1. Can't locate Digest/MD5.pm in @INC (@INC contains: /usr/local/lib64/perl5 /usr/local/share/perl5 /usr/lib64/perl5/vendor_perl /usr/share/perl5/vendor_perl /usr/lib64/perl5 /usr/share/perl5 .) at /usr/local/bin/pt-archiver line 5410.
2. BEGIN failed--compilation aborted at /usr/local/bin/pt-archiver line 5410.

复制代码

办理方案:

1. yum -y install perl-Digest-MD5

复制代码

B.报错如下:

1. Cannot connect to MySQL because the Perl DBI module is not installed or not found.  Run 'perl -MDBI' to see the directories that Perl searches for DBI.  If DBI is not installed, try:
2.   Debian/Ubuntu  apt-get install libdbi-perl
3.   RHEL/CentOS    yum install perl-DBI
4.   OpenSolaris    pkg install pkg:/SUNWpmdbi

复制代码

办理方案:

1. yum install perl-DBI

复制代码

C.报错如下:

1. Cannot connect to MySQL because the Perl DBD::mysql module is not installed or not found.  Run 'perl -MDBD::mysql' to see the directories that Perl searches for DBD::mysql.  If DBD::mysql is not installed, try:
2.   Debian/Ubuntu  apt-get install libdbd-mysql-perl
3.   RHEL/CentOS    yum install perl-DBD-MySQL
4.   OpenSolaris    pgk install pkg:/SUNWapu13dbd-mysql

复制代码

办理方案：

1. yum install perl-DBD-MySQL

复制代码

四.功能优化 

为工具mysql_archiver添加只删除不备份的功能，即只删除源库上的数据，而不用迁移。
1.优化设置表db_archive_info的建表语句

1. DROP TABLE IF EXISTS `db_archive_info`;
2. CREATE TABLE `db_archive_info` (
3.   `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '表id',
4.   `server_source` varchar(640) DEFAULT NULL COMMENT '源服务器',
5.   `port_source` varchar(64) DEFAULT NULL COMMENT '源服务器端口',
6.   `user_source` varchar(64) DEFAULT NULL COMMENT '源数据库用户',
7.   `password_source` varchar(128) DEFAULT NULL COMMENT '源数据库密码',
8.   `db_source` varchar(64) DEFAULT NULL COMMENT '源数据库schema',
9.   `table_source` varchar(128) DEFAULT NULL COMMENT '源数据库表',
10.   <strong>`server_dest` </strong><strong>varchar(640) DEFAULT '' COMMENT '目标服务器',
11.   `port_dest` varchar(64) DEFAULT '' COMMENT '目标服务器端口',
12.   `user_dest` varchar(64) DEFAULT '' COMMENT '目标数据库用户',
13.   `password_dest` varchar(128) DEFAULT '' COMMENT '目标数据库密码',
14.   `db_dest` varchar(64) DEFAULT '' COMMENT '目标数据库schema',
15.   `table_dest` varchar(128) DEFAULT '' COMMENT '目标数据库表'</strong><strong>,</strong>
16.   `archive_condition` varchar(1000) DEFAULT NULL COMMENT '数据归档条件',
17.   `last_archive_date` timestamp NULL DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP COMMENT '最近归档时间',
18.   `last_archive_qty` int(10) DEFAULT NULL COMMENT '最近归档数量',
19.   `state` varchar(255) DEFAULT 'A' COMMENT '数据行状态',
20.   `user_created` varchar(128) DEFAULT 'SYS' COMMENT '数据行创建人',
21.   `datetime_created` timestamp NULL DEFAULT CURRENT_TIMESTAMP COMMENT '数据行创建时间',
22.   `user_modified` varchar(128) DEFAULT 'SYS' COMMENT '数据行修改人',
23.   `datetime_modified` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '数据行修改时间',
24.   PRIMARY KEY (`id`)
25. ) ENGINE=InnoDB AUTO_INCREMENT=14 DEFAULT CHARSET=utf8 COMMENT='数据库归档基础信息表';

复制代码

2.修改备份脚本

1. #!/usr/bin/python
2. # -*- coding: UTF-8 -*-
4. import sys
5. import os
6. import time
7. import db_conn
9. # get db connection
10. db = db_conn.db
11. # use cursor
12. cursor = db.cursor()
14. # 获取命令行参数
15. server_source = sys.argv[1]
16. db_source = sys.argv[2]
18. try:
19.     # SQL 查询语句
20.     sql = "select id, server_source, port_source, user_source, password_source, db_source, table_source," \
21.           "server_dest, port_dest, user_dest, password_dest, db_dest, table_dest, archive_condition " \
22.           "from db_archive_info " \
23.           "where server_source = '%s' and db_source = '%s' " % (server_source, db_source)
25.     # 执行SQL语句
26.     cursor.execute(sql)
27.     # 获取所有记录列表
28.     results = cursor.fetchall()
30.     for row in results:
31.         id = row[0]
32.         server_source = row[1]
33.         port_source = row[2]
34.         user_source = row[3]
35.         password_source = row[4]
36.         db_source = row[5]
37.         table_source = row[6]
38.         server_dest = row[7]
39.         port_dest = row[8]
40.         user_dest = row[9]
41.         password_dest = row[10]
42.         db_dest = row[11]
43.         table_dest = row[12]
44.         archive_condition = row[13]
46.         # 归档开始时间
47.         archive_starttime = time.strftime('%Y-%m-%d %H:%M:%S')
49.         # 生成pt-archive命令
50.         ## 兼容只删除不归档功能，
51.         if server_dest != '' and port_dest != '' and port_dest != '':
52.             archive_cmd = "/usr/local/bin/pt-archiver " \
53.                           "--source h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' " \
54.                           "--dest h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' " \
55.                           "--no-check-charset --charset=utf8mb4 --where '%s' --progress 50000 --limit 10000 --txn-size 10000 " \
56.                           "--statistics --purge " % \
57.                           (server_source, port_source, user_source, password_source, db_source, table_source, \
58.                            server_dest, port_dest, user_dest, password_dest, db_dest, table_dest, \
59.                            archive_condition)
61.         elif server_dest == '' and port_dest == '' and port_dest == '':
62.              archive_cmd = "/usr/local/bin/pt-archiver " \
63.                   "--source h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' " \
64.                   "--no-check-charset --charset=utf8mb4 --where '%s' --progress 50000 --limit 10000 --txn-size 10000 " \
65.                   "--statistics --purge " % \
66.                   (server_source, port_source, user_source, password_source, db_source, table_source,archive_condition)
68.         # print archive_cmd
70.         # make a copy of original stdout route
71.         stdout_archive = sys.stdout
72.         # define the log file that receives your log info
73.         log_file = open("/software/python_script/db_archive_%s_%s.log" % (db_source, table_source), "w")
74.         # redirect print output to log file
75.         sys.stdout = log_file
77.         # archive_cmd = os.popen(pt_archive)
78.         with os.popen(archive_cmd) as c:
79.             # with open("db_archive1.log", "r") as c:
80.             archive_log = c.read()
81.         print(archive_log)
83.         # close log file
84.         log_file.close()
85.         # restore the output to initial pattern
86.         sys.stdout = stdout_archive
88.         # 定义归档相关变量
89.         inserted_qty = 0
90.         deleted_qty = 0
91.         # 归档结束时间
92.         archive_endtime = time.strftime('%Y-%m-%d %H:%M:%S')
94.         with open("/software/python_script/db_archive_%s_%s.log" % (db_source, table_source), "r") as f:
95.             for line in f:
96.                 if 'INSERT' in line:
97.                     i = line.index(" ")
98.                     inserted_qty = line[i + 1:]
99.                 elif 'DELETE' in line:
100.                     i = line.index(" ")
101.                     deleted_qty = line[i + 1:]
103.         # 判断归档是否失败
104.         if server_dest != '' and port_dest != '' and port_dest != '':
105.             if inserted_qty == deleted_qty:
106.                 archive_status = 'Y'
107.                 archive_error = ''
108.             else:
109.                 archive_status = 'N'
110.                 archive_error = 'inserted_qty and deleted_qty are not equal'
111.         elif server_dest == '' and port_dest == '' and port_dest == '':
112.              archive_status = 'Y'
113.              archive_error = 'Delete Data OP'
115.         # insert sql
116.         sql_insert = "insert into db_archive_log(server_source, db_source, table_source, server_dest, " \
117.                      "db_dest, table_dest, archive_qty, archive_cmd, archive_log, archive_start, archive_end, " \
118.                      "archive_status, archive_error ) " \
119.                      "values('%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s')" % \
120.                      (server_source, db_source, table_source, server_dest, \
121.                       db_dest, table_dest, inserted_qty, db.escape_string(archive_cmd), archive_log, archive_starttime,
122.                       archive_endtime, \
123.                       archive_status, archive_error)
124.         # exec sql
125.         cursor.execute(sql_insert)
126.         # exec commit
127.         db.commit()
129.         if archive_status == 'Y':
130.             sql_update = "update db_archive_info " \
131.                          "set datetime_modified = '%s', last_archive_date = '%s', last_archive_qty = %s " \
132.                          "where id = %d" % \
133.                          (archive_starttime, archive_endtime, inserted_qty, id)
134.             cursor.execute(sql_update)
135.             # exec commit
136.             db.commit()
138. except  Exception, e:
139.         print str(Exception)
140.         print str(e)

复制代码

3.阐明
修改db_archive_info的建表语句，将默认值null 调解为‘’，主要是我们通过 server_dest == '' and port_dest == '' and port_dest == '' 来判断是归档还是直接删除。
假如按照修改前的默认值NULL，这几列值不维护，则报错。

1. DBI connect('None;host=None;port=None;charset=utf8mb4;mysql_read_default_group=client','None',...) failed: Unknown MySQL server host 'None' (2) at /usr/local/bin/pt-archiver line 2525.

复制代码

五.功能持续优化（异常）

5.1 代码优化

运行过程中，我们发现步伐不能捕获一些异常，例如当源表和目的表的表结构不一致时，步伐没有报错，异常没有记载到log表中。
还有就是try..except的异常也没有记载下来，针对这些异常情况，我们将代码调解优化如下:

1. #!/usr/bin/python
2. # -*- coding: UTF-8 -*-
4. import sys
5. import os
6. import time
7. import db_conn
9. ### add 部署pt执行过程中的问题
10. import subprocess  ###os.popen不能返回error,故需要使用这个方法
11. #### 以下两行代码是为了解决错误:<type 'exceptions.Exception'>  'ascii' codec can't decode byte 0xe6 in position 0: ordinal not in range(128)
12. reload(sys)
13. sys.setdefaultencoding('utf8')
14. ###
16. # get db connection
17. db = db_conn.db
18. # use cursor
19. cursor = db.cursor()
21. # 获取命令行参数
22. server_source = sys.argv[1]
23. db_source = sys.argv[2]
25. try:
26.     # SQL 查询语句
27.     sql = "select id, server_source, port_source, user_source, password_source, db_source, table_source," \
28.           "server_dest, port_dest, user_dest, password_dest, db_dest, table_dest, archive_condition " \
29.           "from db_archive_info " \
30.           "where server_source = '%s' and db_source = '%s' " % (server_source, db_source)
32.     # 执行SQL语句
33.     cursor.execute(sql)
34.     # 获取所有记录列表
35.     results = cursor.fetchall()
37.     for row in results:
38.         id = row[0]
39.         server_source = row[1]
40.         port_source = row[2]
41.         user_source = row[3]
42.         password_source = row[4]
43.         db_source = row[5]
44.         table_source = row[6]
45.         server_dest = row[7]
46.         port_dest = row[8]
47.         user_dest = row[9]
48.         password_dest = row[10]
49.         db_dest = row[11]
50.         table_dest = row[12]
51.         archive_condition = row[13]
53.         # 归档开始时间
54.         archive_starttime = time.strftime('%Y-%m-%d %H:%M:%S')
56.         # 生成pt-archive命令
57.         ## 兼容只删除不归档功能，
58.         if server_dest != '' and port_dest != '' and port_dest != '':
59.             archive_cmd = "/usr/local/bin/pt-archiver " \
60.                           "--source h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' " \
61.                           "--dest h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' " \
62.                           "--no-check-charset --charset=utf8mb4 --where '%s' --progress 50000 --limit 10000 --txn-size 10000 " \
63.                           "--statistics --purge " % \
64.                           (server_source, port_source, user_source, password_source, db_source, table_source, \
65.                            server_dest, port_dest, user_dest, password_dest, db_dest, table_dest, \
66.                            archive_condition)
68.         elif server_dest == '' and port_dest == '' and port_dest == '':
69.              archive_cmd = "/usr/local/bin/pt-archiver " \
70.                   "--source h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' " \
71.                   "--no-check-charset --charset=utf8mb4 --where '%s' --progress 50000 --limit 10000 --txn-size 10000 " \
72.                   "--statistics --purge " % \
73.                   (server_source, port_source, user_source, password_source, db_source, table_source,archive_condition)
75.         # print archive_cmd
77.         # make a copy of original stdout route
78.         stdout_archive = sys.stdout
79.         # define the log file that receives your log info
80.         log_file = open("/software/python_script/db_archive_%s_%s.log" % (db_source, table_source), "w")
81.         # redirect print output to log file
82.         sys.stdout = log_file
84.         #### os.popen 不能访问error,故用subprocess.Popen来替换。修改执行shell命令的方法。
85.         #修改前
86.         ##with os.popen(archive_cmd) as c:
87.         ##    archive_log = c.read()
88.         ## print(archive_log)
89.         #修改后
90.         sp = subprocess.Popen(archive_cmd,shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
91.         sp_out, sp_err = sp.communicate()
92.         archive_log = sp_out + sp_err
93.         print(archive_log)
94.         ##### end
96.         # close log file
97.         log_file.close()
98.         # restore the output to initial pattern
99.         sys.stdout = stdout_archive
100.         ### start 如果出现出现源表和目标表结构不一致的情况，存在数据库中
101.         if 'The following columns exist in' in archive_log and 'but not' in archive_log :
102.           archive_status = 'N'
103.           archive_error = '源表和目标表数据结构不一致'
104.           archive_starttime = time.strftime('%Y-%m-%d %H:%M:%S')
105.           archive_endtime = time.strftime('%Y-%m-%d %H:%M:%S')
106.           sql_insert = "insert into db_archive_log(server_source, db_source, table_source, server_dest, " \
107.                      "db_dest, table_dest,archive_cmd, archive_log, archive_start, archive_end, " \
108.                      "archive_status, archive_error ) " \
109.                      "values('%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s')" % \
110.                      (server_source, db_source, table_source, server_dest, \
111.                       db_dest, table_dest, cursor._connection.converter.escape(archive_cmd), archive_log, archive_starttime,
112.                       archive_endtime, \
113.                       archive_status, archive_error)
114.           # exec sql
115.           cursor.execute(sql_insert)
116.           # exec commit
117.           db.commit()
119.           continue  ###跳出本循环,开始下一个表 ####其实认为这个continue 可有可无，因为不是不错。
121.         ######  end  异常插入表结束
122.         # 定义归档相关变量
123.         inserted_qty = 0
124.         deleted_qty = 0
125.         # 归档结束时间
126.         archive_endtime = time.strftime('%Y-%m-%d %H:%M:%S')
128.         with open("/software/python_script/db_archive_%s_%s.log" % (db_source, table_source), "r") as f:
129.             for line in f:
130.                 if 'INSERT' in line:
131.                     i = line.index(" ")
132.                     inserted_qty = line[i + 1:]
133.                 elif 'DELETE' in line:
134.                     i = line.index(" ")
135.                     deleted_qty = line[i + 1:]
137.         # 判断归档是否失败
138.         if server_dest != '' and port_dest != '' and port_dest != '':
139.             if inserted_qty == deleted_qty:
140.                 archive_status = 'Y'
141.                 archive_error = ''
142.             else:
143.                 archive_status = 'N'
144.                 archive_error = 'inserted_qty and deleted_qty are not equal'
145.         elif server_dest == '' and port_dest == '' and port_dest == '':
146.              archive_status = 'Y'
147.              ###移除（不归档只删除，没有inserted_qty，，只有deleted_qty，因此赋值）
148.              inserted_qty = deleted_qty
149.              archive_error = 'Delete Data OP'
151.         # insert sql
152.         sql_insert = "insert into db_archive_log(server_source, db_source, table_source, server_dest, " \
153.                      "db_dest, table_dest, archive_qty, archive_cmd, archive_log, archive_start, archive_end, " \
154.                      "archive_status, archive_error ) " \
155.                      "values('%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s','%s')" % \
156.                      (server_source, db_source, table_source, server_dest, \
157.                       db_dest, table_dest, inserted_qty, cursor._connection.converter.escape(archive_cmd), archive_log, archive_starttime,
158.                       archive_endtime, \
159.                       archive_status, archive_error)
160.         # exec sql
161.         cursor.execute(sql_insert)
162.         # exec commit
163.         db.commit()
165.         if archive_status == 'Y':
166.             sql_update = "update db_archive_info " \
167.                          "set datetime_modified = '%s', last_archive_date = '%s', last_archive_qty = %s " \
168.                          "where id = %d" % \
169.                          (archive_starttime, archive_endtime, inserted_qty, id)
170.             cursor.execute(sql_update)
171.             # exec commit
172.             db.commit()
174. except  Exception, e:
175.     ###修改前，print() 一定要放到最后，否则print()后面的代码都不执行了
176.     #print str(Exception)
177.     #print str(e)
178.     #### start 部分错误信息没有记录在log文件中，try...catch 捕获到的错误在此记录,例如:ValueError: I/O operation on closed file。
179.     archive_status = 'N'
180.     archive_error = str(e)
181.     ##print(archive_error)
182.     archive_starttime = time.strftime('%Y-%m-%d %H:%M:%S')
183.     archive_endtime = time.strftime('%Y-%m-%d %H:%M:%S')
184.     sql_insert = "insert into db_archive_log(server_source, db_source,archive_start, archive_end,archive_status, archive_error ) " \
185.                   "values('%s','%s','%s','%s','%s','%s')" % \
186.                   (server_source, db_source, archive_starttime,archive_endtime,archive_status, archive_error)
187.     cursor.execute(sql_insert)
188.     # exec commit
189.     db.commit()
190.     print str(Exception)
191.     print str(e)
192.     #### end

复制代码

5.2 增补阐明

阐明1.
popen 也时引用了subprocess方法，但是 驱动shell命令，没有接受error,只有stdout,我们查看源码。如下:

1. def popen(cmd, mode="r", buffering=-1):
2.     if not isinstance(cmd, str):
3.         raise TypeError("invalid cmd type (%s, expected string)" % type(cmd))
4.     if mode not in ("r", "w"):
5.         raise ValueError("invalid mode %r" % mode)
6.     if buffering == 0 or buffering is None:
7.         raise ValueError("popen() does not support unbuffered streams")
8.     import subprocess, io
9.     if mode == "r":
10.         proc = subprocess.Popen(cmd,
11.                                 shell=True,
12.                                 stdout=subprocess.PIPE,
13.                                 bufsize=buffering)
14.         return _wrap_close(io.TextIOWrapper(proc.stdout), proc)

复制代码

阐明2.
假如源表和目的表，表结构不一致，pt-archiver 报错信息大概不完全相同。
例如:

1. The following columns exist in --dest but not --source:XXXXXXXXXXXXXXXX

复制代码

1. The following columns exist in --source but not --dest:XXXXXXXXXXXXXXXX

复制代码

同样的问题，大概报错不一样。
原因是大概是以源表区比力目的表，也大概是以目的表去反向比力源表。
六.主要参考资料

1.MySQL数据归档实战-Python和pt-archiver的强强结合
https://blog.csdn.net/n88lpo/article/details/78905528?utm_medium=distribute.pc_relevant.none-task-blog-baidulandingword-14&spm=1001.2101.3001.4242
2.MySQL数据归档小工具
https://github.com/dbarun/mysql_archiver#readme
3.在linux下安装MySQLdb及基本利用
https://www.cnblogs.com/blogsme/archive/2012/12/12/2814588.html
4.pt工具之pt-archiver
https://www.cnblogs.com/yhq1314/p/10601801.html
5.linux下pip的安装步骤及使用详解
https://www.jb51.net/article/118035.htm
6.Percona-Toolkit 之 pt-archiver 总结
https://www.cnblogs.com/dbabd/p/10721857.html

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。