GeekTop
  • 首页
  • 关于

GeekTop

不止代码

使用 Python 比较 MySQL 主从的差异

使用 Python 比较 MySQL 主从的差异

2015年12月11日 Alex Comments 0 Comment

由于当时生产环境的数据库使用的是 MySQL5.1 而我们的从库使用的是 MySQL5.6,结果出现了两个数据库表的结构不一致的情况 ,但是我们当时没有用立即发现这种差异,导致后来迁移数据失败,主要原因是 MySQL5.1 与 MySQL5.6 的日志结构差异非常大,所以导致后来表结构的差异。

所以我后来写了一个脚本,专门用来检测检测 MySQL 主从的差异。这个脚本主要实现了表的结构的对比,数据行数的对比,以及部分数据的对比。

# 例外的表,不进行比较
out_tables = ('schema_migrations', 'product_category_hierarchies', 'roles_security_param_values', 'users_roles')
class CompareDatabase(object):
def __init__(self, master_info, slave_info):
self.slave_con = mdb.connect(*master_info)
self.master_con = mdb.connect(*slave_info)
self.slave_cur = self.slave_con.cursor()
self.master_cur = self.master_con.cursor()
# 输入参数:无
# 返回结果(tuple(string)):返回当前主从数据库的版本
def show_version(self):
sql = 'SELECT VERSION()'
self.slave_cur.execute(sql)
self.master_cur.execute(sql)
slave_version = self.slave_cur.fetchone()
master_version = self.master_cur.fetchone()
return master_version[0], slave_version[0]
# 输入参数:
# 1、"master" 返回主库的所有表的名称 type: list(string)
# 2、"slave" 返回从库所有的表的名称 type: lsit(string)
# 3、"all" 返回主库和从库所有的表的明称 type: tuple(list)
# 4、默认返回主库所有表的明称,非法参数返回 None
def show_tables(self, scheme='master'):
sql = 'SHOW TABLES'
if scheme == 'master':
self.master_cur.execute(sql)
master_tables = []
rows = self.master_cur.fetchall()
for row in rows:
if row[0] not in out_tables:
master_tables.append(row[0])
return master_tables
elif scheme == 'slave' and row[0]:
self.slave_cur.execute(sql)
rows = self.master_cur.fetchall()
slave_tables = []
for row in rows:
if row[0] not in out_tables:
slave_tables.append(row[0])
return slave_tables
if scheme == 'all':
self.master_cur.execute(sql)
self.slave_cur.execute(sql)
master_tables = []
rows = self.master_cur.fetchall()
for row in rows:
if row[0] not in out_tables:
master_tables.append(row[0])
slave_tables = []
rows = self.slave_cur.fetchall()
for row in rows:
if row[0] not in out_tables:
slave_tables.append(row[0])
return master_tables, slave_tables
# 返回某一个表的行数,默认返回主库
def show_table_row_num(self, table_name, scheme='master'):
sql = 'SELECT COUNT(*) FROM %s' % table_name
if scheme == 'master':
self.master_cur.execute(sql)
data = self.master_cur.fetchone()
return int(data[0])
elif scheme == 'slave':
self.slave_cur.execute(sql)
data = self.slave_cur.fetchone()
return int(data[0])
elif scheme == 'all':
self.master_cur.execute(sql)
self.slave_cur.execute(sql)
master_data = self.master_cur.fetchone()
slave_data = self.slave_cur.fetchone()
return int(master_data[0]), int(slave_data[0])
else:
return None
# 比较两个库之间表的数目和名称是否完全相同
# 相同返回:True  不同返回:False
def compare_db_tables(self):
master_tables, slave_tables = self.show_tables('all')
for table in master_tables:
if table not in slave_tables:
return False
return True
# 显示表的结构 default ('master')
# 输入参数:
# 1、master 返回主库中该表的结构
# 2、slave 返回从库中该表的结构
# 3、all 返回主库和从库中该表的结构
# 4、非法参数返回None
def show_table_structure(self, table_name, scheme='master'):
sql = 'DESC %s' % table_name
if scheme == 'master':
self.master_cur.execute(sql)
master_rows = self.master_cur.fetchall()
return master_rows
elif scheme == 'slave':
self.slave_cur.execute(sql)
slave_rows = self.slave_cur.fetchall()
return slave_rows
elif scheme == 'all':
self.master_cur.execute(sql)
master_rows = self.master_cur.fetchall()
self.slave_cur.execute(sql)
slave_rows = self.slave_cur.fetchall()
return master_rows, slave_rows
else:
return None
# 显示表的字段 default ('master')
# 输入参数:
# 1、master 返回主库中该表的字段
# 2、slave 返回从库中该表的字段
# 3、all 返回主库和从库中该表的字段
# 4、非法参数返回None
def show_table_columns(self, table, scheme='master'):
if scheme == 'master':
master_table_structure = self.show_table_structure(table)
master_column = []
for row in master_table_structure:
master_column.append(row[0])
return master_column
elif scheme == 'slave':
slave_table_structure = self.show_table_structure(table)
slave_column = []
for row in slave_table_structure:
slave_column.append(row[0])
return slave_column
elif scheme == 'all':
master_table_structure = self.show_table_structure(table)
master_column = []
for row in master_table_structure:
master_column.append(row[0])
slave_table_structure = self.show_table_structure(table)
slave_column = []
for row in slave_table_structure:
slave_column.append(row[0])
return master_column, slave_column
else:
return None
# 比较两个表的结构,相同返回True,不同返回False
def compare_table_structure(self, table_name):
sql = 'DESC %s' % table_name
self.slave_cur.execute(sql)
self.master_cur.execute(sql)
master_rows = self.master_cur.fetchall()
slave_rows = self.slave_cur.fetchall()
for row in master_rows:
if row not in slave_rows:
return False
return True
# 比较连个库所有表的结构,相同返回True,不同返回False
def compare_all_tables_structure(self):
if not self.compare_db_tables():
print('Slave and Master DB tables dose not match!')
return False
master_tables, slave_tables = self.show_tables('all')
for table in master_tables:
if not self.compare_table_structure(table):
print('%s table does not match!' % table)
return False
return True
# 返回表的行数
def show_table_rows(self, table_name):
print('show table %s' % table_name)
sql = 'SELECT COUNT(id) FROM %s' % table_name
self.master_cur.execute(sql)
master_result = self.master_cur.fetchone()
num_of_row = int(master_result[0])
return num_of_row
# 随机返回一个表的前500条记录的50个ID
# 如果表的记录不超过50,则全部返回
def show_table_id_random(self, table_name):
num_of_rows = self.show_table_rows(table_name)
sql = 'SELECT id FROM %s ORDER BY id LIMIT 200' % table_name
if num_of_rows >= 50:
self.master_cur.execute(sql)
master_rows = self.master_cur.fetchall()
selected_rows = random.sample(master_rows, 50)
table_id = []
for row in selected_rows:
table_id.append(int(row[0]))
return table_id
else:
sql = 'SELECT id FROM %s' % table_name
self.master_cur.execute(sql)
selected_rows = self.master_cur.fetchall()
table_id = []
for row in selected_rows:
table_id.append(int(row[0]))
return table_id
# 返回一个表的后50条记录
def show_table_id_last(self, table_name):
num_of_rows = self.show_table_rows(table_name)
sql = 'SELECT id FROM %s ORDER BY id desc LIMIT 50' % table_name
if num_of_rows >= 50:
self.slave_cur.execute(sql)
slave_rows = self.slave_cur.fetchall()
table_id = []
for row in slave_rows:
table_id.append(int(row[0]))
return table_id
else:
sql = 'SELECT id FROM %s' % table_id
self.slave_cur.execute(sql)
slave_rows = self.slave_cur.fetchall()
table_id = []
for row in slave_rows:
table_id.append(int(row[0]))
return table_id
return None
# 随机比较两个表的数据
def compare_one_table_data_random(self, table_name):
table_id = self.show_table_id_random(table_name)
print("Process %s" % table_name)
for id in table_id:
sql = 'SELECT * FROM %s WHERE id = %d' % (table_name, id)
self.master_cur.execute(sql)
self.slave_cur.execute(sql)
master_row = self.master_cur.fetchone()
slave_row = self.slave_cur.fetchone()
if master_row != slave_row:
return False
return True
# 随机比较所有表的数据
def compare_all_tables_data_random(self):
all_tables = self.show_tables()
for table in all_tables:
if self.compare_one_table_data_random(table):
print('%s table is OK' % table)
else:
print('%s table can not pass' % table)
def __del__(self):
self.master_con.close()
self.slave_con.close()

还可以添加一些其他的功能,比如输出文档,但是这不是核心功能。我们后来测试对比主从差异非常小,时间在 1s 以内。


数据库
MySQL, Python

Post navigation

NEXT
读《上山.上山.爱》
PREVIOUS
使用 ipset 管理大批量 IP 地址

发表回复 取消回复

您的电子邮箱地址不会被公开。 必填项已用*标注

最近文章

  • 姥姥
  • 设计一个可扩展的用户模型
  • 使用 Apple 的 Keychain 保存 SSH 的 passphase
  • 解决 ABA 问题
  • 关于 macOS 上面部分 emoji 无法显示的问题
  • 这些年我技术栈的变化
  • 搬瓦工、狗云、hostodo、oracle对比测试
  • Google Drive 无法上传文件
  • 使用 socks5 代理 git ssh 协议
  • 到底要不要“润”
  • 可复用的代码
  • 关于疫情的一点点反思
  • 我的风控策略(投资篇)
  • 我的风控策略(生活篇)
  • Spring Boot是如何处理异常的
  • 编码与解码
  • 基于统计的图像目标检索
  • 有限状态机和状态模式
  • API安全设计
  • 用户密码的存储策略

近期评论

  • ǝɔɐǝԀʎzɐɹϽ发表在《可复用的代码》
  • Alex发表在《可复用的代码》
  • ǝɔɐǝԀʎzɐɹϽ发表在《到底要不要“润”》
  • ǝɔɐǝԀʎzɐɹϽ发表在《可复用的代码》
  • ǝɔɐǝԀʎzɐɹϽ发表在《我的风控策略(生活篇)》
  • ǝɔɐǝԀʎzɐɹϽ发表在《如何写出简洁优雅的代码》
  • 张志亮发表在《如何写出简洁优雅的代码》

分类

  • AI (2)
  • Java应用安全之道 (1)
    • 加密与解密 (1)
  • Odoo (2)
  • Python (1)
  • 图像处理 (1)
  • 年鉴 (1)
  • 数据库 (10)
  • 编程 (14)
    • Spring (1)
  • 读书笔记 (2)
  • 运维 (5)
  • 随笔 (10)

归档

  • 2023年1月 (2)
  • 2022年8月 (1)
  • 2022年7月 (4)
  • 2022年6月 (2)
  • 2022年5月 (2)
  • 2022年4月 (3)
  • 2021年10月 (1)
  • 2021年7月 (1)
  • 2021年5月 (1)
  • 2020年11月 (1)
  • 2020年7月 (1)
  • 2020年3月 (2)
  • 2020年2月 (1)
  • 2019年1月 (1)
  • 2018年12月 (2)
  • 2018年11月 (2)
  • 2017年4月 (1)
  • 2016年11月 (1)
  • 2016年9月 (1)
  • 2016年7月 (1)
  • 2016年5月 (3)
  • 2016年4月 (2)
  • 2016年3月 (1)
  • 2016年2月 (2)
  • 2015年12月 (1)
  • 2015年11月 (2)
  • 2015年8月 (1)
  • 2015年4月 (1)
  • 2015年3月 (1)

标签

Database devops Java MySQL PostgreSQL Python shell Spring SpringBoot Spring Security 安全 年鉴 总结 编程 随笔
© 2015-2023   Geektop.net All Rights Reserved.