数据一致性校验（pt-table-checksum）

介绍

• pt-table-checksum 和 pt-table-sync 是 percona 公司发布的、检查 MySQL 主从数据库数据一致性校验的工具。
• pt-table-checksum 利用 MySQL 复制原理，在主库执行校验和计算，并对比主从库校验和，由此判断主从库数据是否一致。
• 如果发现数据不一致，或者命令执行报 warning 或 error，则返回状态值为非0。
• pt-table-checksum 用于找出数据差异，pt-table-sync 用于修复数据差异。
• 即使数据库有几十万张表，数万亿行数据，pt-table-checksum 运行也很良好。

工作原理

• pt-table-checksum 运行时，将每张表分成多个 chunk，并使用 replace ... select 查询，对每个 chunk 进行 checksum 计算。
• 它可以改变 chunk size，以使校验和查询在所需的时间内运行，每个 chunk 的目标时间默认是 0.5 秒。
• 它可以跟踪服务器执行查询的速度，并在服务器性能负载过重时，如流量高峰或后台任务时，自动调节 chunk size。
• 执行校验的表要有主键或唯一索引，这样校验的速度才会快。
• chunk size 可以使用 --chunk-size-limit 设置 chunk 的最大值。
• 运行时，它会设置 session 级的 innodb_lock_wait_time_out 为 1，如果存在锁等待的话，它会释放锁，保证数据库能够正常读写。
• 默认一张表同时有25个查询时，它将暂停。查询数可以使用 --max-load 设置。
• 如果它的查询被 kill 的话，它将重试一次，如果再次失败，它将跳到下一个 chunk。如果锁等待超时，它会执行相同的行为。如果发生此类错误，将打印 warning，但是每张表只会打印一次。如果网络中断，它会尝试重连并继续执行。
• 如果 pt-table-checksum 异常终止了，可以加上 --resume 选项恢复它，它将从上次处理的最后一个 chunk 开始。可以随时执行 CTRL+C 手动停止，并再次恢复。
• 当检查完所有表块的校验和后，它将暂停并等待所有从库执行完校验和查询。随后检查所有的从库与主库的数据是否相同，并打印一行结果。
• 可以手动执行如下查询，查询校验和信息

1 2 3 4 5 6 7

SELECT db, tbl, SUM(this_cnt) AS total_rows, COUNT(*) AS chunks FROM percona.checksums WHERE ( master_cnt <> this_cnt OR master_crc <> this_crc OR ISNULL(master_crc) <> ISNULL(this_crc)) GROUP BY db, tbl;

限制

• pt-table-checksum 需要基于 statement 的复制，运行时需要在 master 上设置 binlog_format=statement，从库不接受此项更改。
• 工具运行时会检查所有库的 binlog 格式。
• 如果复制从库没有主库上的校验和架构、或者主从库的表结构不一致，复制会中断。

退出状态

• 0 为正常，没有 warning、error、校验和差异、或跳过的表、块。
• 255 为严重错误，
• 其他大于 0，小于 255 的值，都表示一种异常，可以查看手册。

选项

选项名	说明	默认值	类型	组别
--ask-pass	执行时手动输入密码
--[no]check-binlog-format	检查所有库的 binlog_format 值	yes
--binary-index	改变 --create-replicate-table 创建的表类型，改成 BLOB 类型，用于字符集非标准字符集时。
--check-interval		1		阈值
--[no]check-plan	工具使用了几种启发式方法判断执行计划是否有错，如果执行计划有错误，将跳过表块，开启会增加执行时间	yes
--[no]check-replication-filters	如果主库或从库设置了binlog_ignore_db 或 replicate_do_db 的过滤选项，因为不能保证查询和校验和不会导致复制中断，所以执行工具会因报错而终止。 使用工具使要确保没有过滤参数，再开启此选项。	yes		安全
--check-slave-lag	工具会监视所有从库的延迟，如果某个从库延迟超过 --max-lag 的值，则暂停执行校验和，直到从库的延迟小于 --max-lag 的值。		string	阈值
--[no]check-slave-tables	检查所有从库的表都存在、且表的列与主库的列相同，避免因为主从库表结构不一致导致复制中断。 同样是在确保主从库表结构一致时，再开启此选项。	yes		安全
--chunk-index	为工具设置 chunk 时强制指定的索引，指定的索引会被工具以 FORCE INDEX 子句写入 SQL 语句中。注意手动选择索引可能导致性能下降。 当默认索引不合适可以手动指定索引，但当手动指定索引不存在，工具将回归默认方式选择索引。		string
--chunk-index-columns	设置复合索引的最左索引列的列数。 复合索引包含列数过多时，可能导致 mysql 的扫描范围过大或执行计划错误，此选项可以提高索引选择效率。 通常在包含4个列或4个以上的列时使用。		int
--chunk-size	为执行每次校验和的SQL设置rows，后缀可以为 k、m、g。通常不用设置此项，设置 --chunk-time 更有效。 设置此项会导致动态调整 chunk 失效，并使每个 chunk 都等于设置的值。另外，如果设置5000，但某个where 子句只能匹配1000，那么这个chunk可能会被跳过。 设置较小会导致校验变慢，因为需要设置 --[no]check-plan	1000	size
--chunk-size-limit	块大小限制倍数。因为工具获取 chunk 的行数，是来自执行计划中估计的行数，而非准确值。 如果该 chunk 超过了 1000 * limit 的大小，工具将跳过该块。 如果 limit 为1，表示任何 chunk 都不能大于 --chunk-size；如果 limit 为0，则表示没有限制。	2.0	float	安全
--chunk-time	动态调整块大小，以便每个校验和查询都需要这么长的时间来执行。如果调整该参数为0，将不会自动调整块大小。	0.5	float
--columns	缩写 -c。如果校验的表不包含指定的列，则跳过该表。 此选项适用于所有表，因此它实际上只在对一个表进行校验和时才有意义，除非这些表有一组公共列。		array	过滤器
--config	读取以逗号分割的配置文件列表，如果配置该选项，该选项必须是第一个参数。		array	配置
--[no]create-replicate-table	如果不存在 replicate 库和表，则创建它们。	yes
--databases	缩写 -d。仅对指定的库列表进行校验和。		hash	过滤器
--databases-regex	只校验 Perl 正则匹配的数据库。		string	过滤器
--defaults-file	从指定的文件读取 mysql 参数。该文件必须包含绝对路径。		string	连接
--[no]empty-replicate-table	对表进行校验和之前，删除复制表之前的校验和结果。删除不是执行 truncate，而是检查到对应表时再删除对应结果。 表不会自动清空，为了下次执行时可以使用之前的结果。可以手动清空。	yes
--engines	缩写 -e。只检查指定引擎的表。		hash	过滤器
--explain	显示计划，但不执行查询、校验和。如果指定2次，则打印每个 chunk 的上下边界，但不执行查询、校验和。	0	累计值	输出
--float-precision	浮点和双数到字符串转换的精度。使用MySQL中的ROUND() 函数将浮点值和双精度值舍入到小数点后指定的位数。		int
--function	校验和的 hash 函数（FNV1A_64、murtur_Hash、SHA1、MD5、CRC32等）默认是 crc32，也可以自己定义，但是函数必须是 mysql 内部支持的。mysql 内部没有很好的哈希函数，crc32 可能会有冲突，但是 md5 和 sha1 会非常占用 cpu。		string
--host	缩写 -h。默认 localhost。		string	连接
--ignore-columns	计算校验和时忽略此以逗号分隔的列列表。如果表的所有列都按–忽略列筛选，则将跳过该表。		hash	过滤器
--ignore-databases	忽略此以逗号分隔的数据库列表。		hash	过滤器
--ignore-databases-regex	忽略名称与此Perl正则表达式匹配的数据库。		string	过滤器
--ignore-engines	忽略此以逗号分隔的存储引擎列表。	FEDERATED,MRG_MyISAM	hash	过滤器
--ignore-tables	忽略这个用逗号分隔的表列表。		hash	过滤器
--ignore-tables-regex	忽略名称与Perl regex匹配的表。		string	过滤器
--max-lag	暂停校验和，直到所有副本的延迟小于此值。工具使用 Seconds_Behind_Master 来比较。如果任何从库的延迟时间超过这个值，则工具将休眠 --check-interval 秒。可以指定 --check-slave-lag，则只检查指定服务器的 lag，而不是所有从库。	1s	time	阈值
--max-load	工具检查每个块之后，执行 show global status，如果任何状态变量高于阈值，则暂停。该选项接受以逗号分隔的MySQL状态变量列表来检查阈值。每个变量后面都可以有一个可选的=MAX_VALUE。如果没有给出，工具通过检查当前值并将其增加20%来确定阈值。如 Threads_running=100，高于120时暂停，小于120时恢复运行。	Threads_running=25	array	阈值
--password	缩写 -p。连接时使用的密码。		string	连接
--pause-file	当此参数指定的文件存在时，将暂停执行。		string
--pid	创建指定的 pid 文件。如果pid 文件已经存在，并且它包含的 pid 与当前 pid 不同，则该工具不会启动。 但是，如果 pid 文件存在并且它包含的 pid 不再运行，则该工具将用当前 pid 覆盖该 pid 文件。当工具退出时，将自动删除 pid 文件。		string
--plugin	定义一个 pt_table_checksum_plugin 类的模块。perl 模块可以挂载到工具的很多部分中。		string
--port	缩写 -P。连接的端口号。		int	连接
--progress	将进度报告打印到STDERR。该值是一个逗号分隔的列表，由两部分组成。第一部分可以是百分比、时间或迭代；第二部分指定更新的打印频率（以百分比、秒或迭代次数为单位）。	time,30	array
--quiet	缩写 -q。只打印重要信息。指定此选项一次将导致工具仅打印具有校验和差异的错误、警告和表。	0	累计值
--recurse	发现从库时在层次结构中递归的级别数。默认值是无限的。		int
--recursion-method	发现从库的递归方法。尽管运行工具不需要检查从库，但是工具不能检查未发现从库中的差异。 可选值还有 cluster、dsn=DSN 、none。 dsn 比较特殊，需要创建特定表并写入数据库信息。	processlist,hosts	array
--replicate	将校验和结果写入指定表。 默认情况下，--[no]create replicate table为true，因此 如果选项不存在，则会自动创建它们。	percona.checksums	string
--[no]replicate-check	完成每个表后，检查副本中的数据差异。	yes
--replicate-check-only	在不执行校验和查询的情况下检查从库的一致性。
--replicate-check-retries	遇到差异时，重试校验和比较的次数。只有当此检查次数之后差异仍然存在时，才认为它有效。将此选项的值设置为2或更大可以缓解使用 –resume 选项时出现的虚假差异。	1	int
--replicate-database	只使用指定数据库存放校验结果表。		string
--resume	从最后完成的块恢复校验和。
--retries	当出现非致命错误时，请多次重试块。非致命错误是诸如锁等待超时或查询被终止等问题。	2	int
--run-time	设置执行校验和的总时间。默认执行完所有库表的校验和后终止。单位后缀有：s、m、h、d。 通常与 --resume 结合使用，可以在分配时间内对尽可能多的表做校验和。		time
--separator	用于 CONCAT_WS() 的分隔符。此字符用于在校验和时联接列的值。	#	string
--skip-check-slave-lag	跳过检查某个从库的延迟，后面填写从库的 DSN。 可以重复，每个从库前面要加一次该选项。		DSN
--slave-user	设置用于连接到从库的用户。		string
--slave-password	设置用于连接从属服务器的密码。它要与–slave user一起使用，并且所有slave上的用户密码必须相同。		string
--set-vars	在 var=value 对的逗号分割列表中设置 mysql 变量。如果无法设置变量，工具将打印警告并继续。		array	连接
--socket	缩写 -S。socket 文件。		string	连接
--slave-skip-tolerance	当主表被标记为仅在一个块中进行校验sum，但从表超过了该块的最大可接受大小时，将跳过该表。	1.0	float
--tables	缩写 -t。仅校验以逗号分隔的表列表的校验和。表名可以用数据库名限定。		hash	过滤器
--tables-regex	仅对名称与此Perl正则表达式匹配的表进行校验和。		string	过滤器
--trim	将 trim() 添加到 varchar 列。
--user	缩写 -u。登陆用户		string	连接
--[no]version-check	检查工具的版本号。		yes
--where	匹配指定的行，只校验部分数据

python 自动一致性校验脚本

db_consistent_check.py 折叠源码

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 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225

#!/usr/bin/env python3 # -*- coding: UTF-8 -*- '''================================================= @Project -> File   ：db_consistent_check -> db_consistent_check @IDE    ：PyCharm @Author ：DingLu @Date   ：2020-02-24 12:27 @Desc   ：mysql databases data consistent check ==================================================''' """ 检查 ip port 是否是从库     如果是从库，则获取主库 ip port 检查 ip port 不是从库     如果是单实例         break     如果是主库         pass """ """ 检查主从库 port 是否一致     一致         --recursion-method="processlist"     不一致         --recursion-method="hosts"         或         --recursion-method=dsn=D=xxx,t=xxx """ """ 检查数据库参数     binlog_format 是否是 row         是             添加 --nocheck-binlog-format 选项         不是             pass     replication_do_db 空         --nocheck-replication-filters """ """ 是否指定校验和存储表     是         --replicate=xxx     否         不添加 """ """ 是否指定数据库     是         --databases=xxx     否         不添加 """ """ 是否指定表     是         --tables         或         --tables-regex     否         不添加 """ """ 拼接命令     h     u     p     P       """ import os import sys from lib.ssh_server import * from lib.connect_MySQL import * from lib.get_local_ip_addr import * project_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.append(os.getcwd().split(project_dir)[0] + project_dir) class Db_Consis_Check(object):     """     数据一致性校验     """     def __init__(self, dic):         self.dic = dic         self.host = dic['host']         self.port = dic['port']         self.user = dic['myuser']         self.pwd  = dic['mypwd']     def con_db(self):         """         连接 db         :return: conn 对象         """         db_conn = ConnectMySQL(self.dic)         return db_conn     def is_slave(self):         """         是否为从库         :return: host_dic         """         sql = "show slave status;"         slave_info_dic_list = self.con_db().execute_sql(sql)         if slave_info_dic_list:             slave_info_dic = slave_info_dic_list[0]             if slave_info_dic:                 master_ip = slave_info_dic['Master_Host']                 master_port = slave_info_dic['Master_Port']                 self.dic['host'] = master_ip                 self.dic['port'] = master_port             print(self.dic, "是从库")         return self.dic     def is_master(self):         """         是否为主库         :return: host_dic         """         sql = "show slave hosts;"         slave_hosts_dic_list = self.con_db().execute_sql(sql)         if not slave_hosts_dic_list:             print(self.host, "没有从库, 不能运行 pt-table-checksum")             return None         if slave_hosts_dic_list:             print(self.host, "是主库")         # 判断从库端口和主库是否一致，如果不一致需要添加 --recursion-method="hosts"         for dic in slave_hosts_dic_list:             if int(dic['Port']) != int(self.port):                 print("从库", dic['Server_id'] , '与主库端口不一致，需要添加 --recursion-method="hosts"')                 break             else:                 pass         print("从库与主库端口一致")         return self.dic     def is_row_format(self):         sql = "show variables like 'binlog_format';"         format_dic_list = self.con_db().execute_sql(sql)         format_dic = format_dic_list[0]         if format_dic['Variable_name'] == 'binlog_format' and format_dic['Value'] == 'ROW':             print("行格式是 ROW，需要添加 --nocheck-binlog-format 选项")             return "--nocheck-binlog-format"         else:             return None     def is_repl_filter(self, repl_filter_flag):         """         是否有复制过滤         :param repl_filter_flag: 过滤标志         :return: --nocheck-replication-filters         """         if repl_filter_flag:             opt = "--check-replication-filters"         if not repl_filter_flag:             opt = "--nocheck-replication-filters"         print("没有过滤参数，可以添加 ", opt, "选项")         return opt     def allocate_dbs(self, dbs_name_string):         """         指定校验db         :param dbs_name_string: db1,db2,dbn         :return: --databases=db1,db2,dbn         """         opt = "--databases=" + dbs_name_string         print(opt)         return opt     def allocate_table(self, db_name_string, tables_name_string):         """         指定表，一个或多个表         :param db_name_string: db1         :param tables_name_string: tb1,tb2,tbn         :return: --databases=db1 --tables=tb1,tb2,tbn         """         if "," not in db_name_string:             # 指定了单个db             opt = "--databases=" + db_name_string + ' ' + "--tables=" + tables_name_string             return opt         elif "," in db_name_string:             print("检查指定表时，只能指定单个DB")             return None     def merge_cmd(self, *args, **kwargs):         # 拼接命令         opts = opt1 + " " + opt2 + " " + opt3 + " " + opt4         cmd_string = "pt-table_checksum " + opts + "h=" + self.host + ",u=" + self.user + ",p=" + self.pwd + ",P=" + str(self.port)         print(cmd_string)         return cmd_string     def run_cmd(self,cmd):         # 运行命令         local_ip = get_local_ip()         if self.host == local_ip:             # 本地执行             stdout = os.system(cmd)             return stdout         elif self.host != local_ip:             # 远程执行             obj2 = SshServer(self.dic, cmd)             remote_stdout = obj2.ssh_to_server()             return remote_stdout if __name__ == '__main__':     host_dic = {'host': '172.16.250.29',                 'port': 3306,                 'uname': 'xxx',                 'pwd': 'xxx',                 'myuser': 'xxx',                 'mypwd': 'xxx'}     obj = Db_Consis_Check(host_dic)     host_dic = obj.is_slave()     host_dic2 = obj.is_master()     if not host_dic2:         opt1 = obj.is_row_format()         opt2 = obj.is_repl_filter('')         opt3 = opt4 = ''         cmd = obj.merge_cmd(opt1, opt2, opt3, opt4)         obj.run_cmd(cmd)