使用python实现MySQL表结构比对【初级版】
2023-06-13 09:17:28 时间
使用Python写了个最基本的表结构比对告警脚本, 目前这个版本只能做到发现表结构存在差异的表。后续考虑再迭代下,看如何把差异的表和列的明细也报出来。
configs.py 内容如下:
# MySQL数据源的信息
mysql_source_host = "localhost"
mysql_source_port = "3306"
mysql_source_user = "dts"
mysql_source_pass = "dts"
mysql_source_db = "src_db"
# MySQL目标库的信息
mysql_dest_host = "localhost"
mysql_dest_port = "3306"
mysql_dest_user = "dts"
mysql_dest_pass = "dts"
mysql_dest_db = "dest_db"
main.py 内容如下:
# pip3 install mysql-connector-python==8.0.31
import mysql.connector
import time
import configs
import hashlib
start_time = time.time()
# 源端
source_db = mysql.connector.connect(
host=configs.mysql_source_host,
port=configs.mysql_source_port,
user=configs.mysql_source_user,
passwd=configs.mysql_source_pass,
)
source_cursor = source_db.cursor()
# 目标端
dest_db = mysql.connector.connect(
host=configs.mysql_dest_host,
port=configs.mysql_dest_port,
user=configs.mysql_dest_user,
passwd=configs.mysql_dest_pass,
)
dest_cursor = dest_db.cursor()
# 1 获取源端的表清单
get_src_tb_list = (
"SELECT table_name FROM information_schema.tables WHERE table_schema="
+ "'" + configs.mysql_source_db + "'"
)
source_cursor.execute(get_src_tb_list)
src_tb_result = source_cursor.fetchall()
src_tb_set = set()
for x in src_tb_result:
tb_name = x[0]
src_tb_set.add(tb_name)
print(f"源端的表明细:", src_tb_set)
# 2 获取目标端的表清单
get_dest_tb_list = (
"SELECT table_name FROM information_schema.tables WHERE table_schema="
+ "'" + configs.mysql_dest_db + "'"
)
dest_cursor.execute(get_dest_tb_list)
dest_tb_result = dest_cursor.fetchall()
dest_tb_set = set()
for x in dest_tb_result:
tb_name = x[0]
dest_tb_set.add(tb_name)
print(f"目标端的表明细:", dest_tb_set)
print("----------- 计算源端和目标端的差集 --------------------")
print(f"dest比src多的表:", list(dest_tb_set - src_tb_set))
print(f"src比dest多的表:", list(src_tb_set - dest_tb_set))
# print("------------ 开始比较每个表的每个列信息 -------------------")
base_sql = '''
SELECT
table_name,
column_name,
CASE WHEN `COLUMN_DEFAULT` IS NULL THEN '' ELSE `COLUMN_DEFAULT` END AS `COLUMN_DEFAULT` ,
IS_NULLABLE,
COLUMN_TYPE,
COLUMN_KEY,
EXTRA
FROM
information_schema.columns where table_schema =
'''
source_chksum = dict()
dest_chksum = dict()
# 3 采集源库数据
for i in src_tb_set:
get_src_tb_column_detail = (
base_sql
+ "'" + configs.mysql_source_db + "'" + "AND table_name=" + "'" + str(
i) + "' ORDER BY ordinal_position ASC;"
)
source_cursor.execute(get_src_tb_column_detail)
src_tb_column_result = source_cursor.fetchall()
chk_sum = hashlib.md5(str(src_tb_column_result).replace(" ", "").encode()).hexdigest()
source_chksum[i] = chk_sum
# 4 采集目标库信息
for i in dest_tb_set:
get_dest_tb_column_detail = (
base_sql
+ "'" + configs.mysql_dest_db + "'" + "AND table_name=" + "'" + str(i) + "' ORDER BY ordinal_position ASC;"
)
dest_cursor.execute(get_dest_tb_column_detail)
dest_tb_column_result = dest_cursor.fetchall()
chk_sum = hashlib.md5(str(dest_tb_column_result).replace(" ", "").encode()).hexdigest()
dest_chksum[i] = chk_sum
# 5 进行集合运算
print("-------------- 结果统计 ------------------")
if source_chksum != dest_chksum:
s1 = set()
differ = set(source_chksum.items()) ^ set(dest_chksum.items())
for i in differ:
# 如果要排除掉src和dest 存在差集的表(背景:有时候源库src已经建好表,但是尚未发布到生产dest去,这种情况下就出现了二者表的数量不一样多),用下面这种写法
# if i[0] not in list(dest_tb_set - src_tb_set ) and i not in list(src_tb_set - dest_tb_set):
# s1.add(i[0])
# 如果要全部都报出来,用下面这种写法
s1.add(i[0])
print('表结构(只比对了列和索引,未包括字符集和排序集)检查完成,存在差异的表如下 ---> ', s1)
else:
print('表结构(只比对了列和索引,未包括字符集和排序集)检查完成,没有发现存在差异的表')
stop_time = time.time()
time_dur = stop_time - start_time
print(f"耗时 {time_dur} 秒")
readme 内容如下:
上面程序里面,我们没有比对表中的字符集和排序集。
如果要比较字符集使用下面的:
base_sql = '''
SELECT
table_name,
column_name,
CASE WHEN `COLUMN_DEFAULT` IS NULL THEN '' ELSE `COLUMN_DEFAULT` END AS `COLUMN_DEFAULT` ,
IS_NULLABLE,
COLUMN_TYPE,
CASE WHEN CHARACTER_SET_NAME IS NULL THEN '' ELSE CHARACTER_SET_NAME END AS CHARACTER_SET_NAME ,
CASE WHEN COLLATION_NAME IS NULL THEN '' ELSE COLLATION_NAME END AS COLLATION_NAME ,
COLUMN_KEY,
EXTRA
FROM
information_schema.columns where table_schema =
'''
如果不比较字符集,使用下面的:
base_sql = '''
SELECT
table_name,
column_name,
CASE WHEN `COLUMN_DEFAULT` IS NULL THEN '' ELSE `COLUMN_DEFAULT` END AS `COLUMN_DEFAULT` ,
IS_NULLABLE,
COLUMN_TYPE,
COLUMN_KEY,
EXTRA
FROM
information_schema.columns where table_schema =
'''
使用方法: 修改完成 configs.py后,执行 python3 main.py 即可
相关文章
- Python进阶42-drf框架(四)
- Python赋值运算符(入门必读)
- MySQL 构建高可用集群,实现卓越数据稳定性(mysql集群高可用)
- MySQL中存储大量文本数据的解决方案(mysql长文本类型)
- MySQL 日期比较:从小到大(mysql日期比较大小)
- Mysql ID自增长技术实现畅通无阻(mysql的id自增长)
- MySQL表结构同步的技术实现(mysql同步表结构)
- MySQL字段类型转换:实现正确的数据格式(mysql字段类型转换)
- MySQL数据:实现导出与导入(mysql数据导出导入)
- MySQL实现修改表的字符编码(mysql修改表的编码)
- 轻松使用MySQL实现自动建表(mysql自动建表)
- MySQL连接池:实现优化性能的配置实践(mysql连接池配置)
- MySQL中实现条件查询的有效语句(mysql中条件查询语句)
- MySQL设置默认为当前时间的方法(mysql默认值当前时间)
- MySQL中使用IF语句的基础知识(mysql的if语句)
- MySQL连接服务器的简易步骤(mysql如何连接服务器)
- MySQL实现自动生成随机字符串(mysql生成随机字符串)
- MySQL 创建表之主键制定法(mysql创建表主键)
- MySQL官方网站无法下载(mysql官网不能下载)
- MySQL实现降序查询的技巧(mysql降序查询)
- Mysql实现List存储的技巧(mysql存储list)
- MySQL中查询时间格式的正确方式(mysql查询时间格式)
- 优化数据库查询效率:使用MySQL大于符号索引实现快速查询(mysql大于索引)
- MySQL快速实现汉字首字母索引(mysql取汉字首字母)
- 如何在 MySQL 中复制一列数据(mysql复制一列数据)
- 防范MySQL木马攻击:有效方法分享(mysql木马)
- 使用Mul MySQL实现高效数据管理与存储(mul mysql)
- MySQL数据库双活实现数据安全最佳保障(mysql数据库双活)
- MySQL的MVCC技术:实现数据库性能的优化篇(mvcc mysql)
- MySQL数据库集群:实现高可用性的搭建实践(mysql数据库集群搭建)
- MySQL:探索所有表的秘密(mysql 所有表)
- WPF连接MySQL:实现步骤深度剖析(wpf 连接mysql)
- MySQL实现两行相加功能,轻松拼接数据(mysql中两行相加)
- CSV与织梦MySQL实现数据迁移的简单操作(csv织梦mysql)
- 在CM中配置Mysql使用指南(cm中使用mysql)
- MySQL数据库的分区和分片功能优化查询和提高性能(mysql中分区和分片)
- MySQL语句快速不查询指定列(mysql不查询指定列)