用python将几个mysql数据库的数据同步到一个mysql里面

Posted 2023-05-02

现有几个分厂的mysql数据库数据，需要定期同步到总厂的mysql数据库中，用python编写，请帮忙解答一下

MySQL 的 Binlog 记录着 MySQL 数据库的所有变更信息，了解 Binlog 的结构可以帮助我们解析Binlog，甚至对 Binlog 进行一些修改，或者说是“篡改”，例如实现类似于 Oracle 的 flashback 的功能，恢复误删除的记录，把 update 的记录再还原回去等。本文将带您探讨一下这些神奇功能的实现，您会发现比您想象地要简单得多。本文指的 Binlog 是 ROW 模式的 Binlog，这也是 MySQL 8 里的默认模式，STATEMENT 模式因为使用中有很多限制，现在用得越来越少了。
Binlog 由事件（event）组成，请注意是事件（event）不是事务（transaction），一个事务可以包含多个事件。事件描述对数据库的修改内容。
现在我们已经了解了 Binlog 的结构，我们可以试着修改 Binlog 里的数据。例如前面举例的 Binlog 删除了一条记录，我们可以试着把这条记录恢复，Binlog 里面有个删除行（DELETE_ROWS_EVENT）的事件，就是这个事件删除了记录，这个事件和写行（WRITE_ROWS_EVENT）的事件的数据结构是完全一样的，只是删除行事件的类型是 32，写行事件的类型是 30，我们把对应的 Binlog 位置的 32 改成 30 即可把已经删除的记录再插入回去。从前面的 “show binlog events” 里面可看到这个 DELETE_ROWS_EVENT 是从位置 378 开始的，这里的位置就是 Binlog 文件的实际位置（以字节为单位）。从事件（event）的结构里面可以看到 type_code 是在 event 的第 5 个字节，我们写个 Python 小程序把把第383（378+5=383）字节改成 30 即可。当然您也可以用二进制编辑工具来改。
找出 Binlog 中的大事务
由于 ROW 模式的 Binlog 是每一个变更都记录一条日志，因此一个简单的 SQL，在 Binlog 里可能会产生一个巨无霸的事务，例如一个不带 where 的 update 或 delete 语句，修改了全表里面的所有记录，每条记录都在 Binlog 里面记录一次，结果是一个巨大的事务记录。这样的大事务经常是产生麻烦的根源。我的一个客户有一次向我抱怨，一个 Binlog 前滚，滚了两天也没有动静，我把那个 Binlog 解析了一下，发现里面有个事务产生了 1.4G 的记录，修改了 66 万条记录！下面是一个简单的找出 Binlog 中大事务的 Python 小程序，我们知道用 mysqlbinlog 解析的 Binlog，每个事务都是以 BEGIN 开头，以 COMMIT 结束。我们找出 BENGIN 前面的 “# at” 的位置，检查 COMMIT 后面的 “# at” 位置，这两个位置相减即可计算出这个事务的大小，下面是这个 Python 程序的例子。
切割 Binlog 中的大事务
对于大的事务，MySQL 会把它分解成多个事件（注意一个是事务 TRANSACTION，另一个是事件 EVENT），事件的大小由参数 binlog-row-event-max-size 决定，这个参数默认是 8K。因此我们可以把若干个事件切割成一个单独的略小的事务
ROW 模式下，即使我们只更新了一条记录的其中某个字段，也会记录每个字段变更前后的值，这个行为是 binlog_row_image 参数控制的，这个参数有 3 个值，默认为 FULL，也就是记录列的所有修改，即使字段没有发生变更也会记录。这样我们就可以实现类似 Oracle 的 flashback 的功能，我个人估计 MySQL 未来的版本从可能会基于 Binlog 推出这样的功能。
了解了 Binlog 的结构，再加上 Python 这把瑞士军刀，我们还可以实现很多功能，例如我们可以统计哪个表被修改地最多？我们还可以把 Binlog 切割成一段一段的，然后再重组，可以灵活地进行 MySQL 数据库的修改和迁移等工作。 参考技术A 这是我以前写的一个简单的查找xml值的一个方法，使用该方法可以比较方便的获得xml的值，相比xml模块来说比较轻巧defgetValue(xmlcontent,type):start=''end=''ifxmlcontent.find(start)==-1orxmlcontent.find(end)==-1:return""else:sid=xmlcontent.find(start)+len(type)+2eid=xmlcontent.find(end)returnxmlcontent[sid:eid]例如:a='''Jim12'''name=getValue(a,'name')age=getValue(a,'age')在这获得了数据之后再写到数据库不过是一条简单的INSERT语句罢了追问

可以在详细讲一下吗

本回答被提问者和网友采纳

MySQL 5.7 多主一丛同步的数据库配置（将多处数据源合并到一点）

工作需要，笔记之用。文章很长，倒一杯茶慢慢看。

 

数据库的应用场景颇多，如 数据库双机同步，一主多从，多主多从，多主一从等；下文记录多主一从的配置及测试。

大多数复制场景中是一主或者一主多从。这种拓扑用于高可用性场景，读写分离。主机负责写入数据，丛集负责读数据，横向扩展读取程序。但是，多主一从是写入多个数据库实例，最后合并成一个结果。

多主一从使得从机从各主机同步接收业务信息（transactions），这样可以一部服务器为多个主机服务器备份，合并数据表，联合数据。（无去重）

 

　　　　

 

MySQL 版本：5.7.10

配置机器：两主一从

1，从机配置，保证从机的仓库都存在一个表；

[mysqld]
master-info-repository=TABLE
relay-log-info-repository=TABLE

确保无误后，可以检查一下，如下显示则配置正常：

mysql> SHOW VARIABLES LIKE ‘%repository%‘;
+---------------------------+-------+
| Variable_name | Value |
+---------------------------+-------+
| master_info_repository | TABLE |
| relay_log_info_repository | TABLE |
+---------------------------+-------+
2 rows in set (0.00 sec)

接下来修改 server_id ，这里用了ip的后三位：

[mysqld]
server-id=141

输入下面代码可以捡测：

mysql> SHOW VARIABLES WHERE VARIABLE_NAME = ‘server_id‘;
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| server_id | 141 |
+---------------+-------+
1 row in set (0.00 sec)　　

 

---

 注意：MySQL 5.7 安装后，请在错误日志里面去找密码，如下：

# grep root error.log
2015-12-09T05:34:01.639797Z 1 [Note] A temporary password is generated for [email protected]: T<e-hd0cgI!d　　

 

接下来你肯定需要用到这个密码，所以最好修改一下，示例改为了“password”：

ALTER USER ‘root‘@‘localhost‘ IDENTIFIED BY ‘password‘;</font>

　　

 多主一从最关键的工作就是保证在你的两（或多）个源数据里面不要有相同的主键，特别是当你在用AUTO_INCREMENT 这一列时，如果有两个一样的主键可以想象同步到从机时数据就会紊乱。这里有一个可替代的方法作为参考。在配置时最好关掉GTID，不然在mysql initialize 初始化时会遇到一个问题，并且这些记录会被复制到从机上去。假设你最开始启动了GTID，配置完成后，然后你试图在从机上复制。当你检查主机的状态时（输入 SHOW MASTER STATUS）， 你会看到这样的结果，主机上138个执行记录，现在会被复制到从机上：

mysql> SHOW MASTER STATUS\\G
*************************** 1. row ***************************
File: mysql-bin.000002
Position: 1286
Binlog_Do_DB:
Binlog_Ignore_DB:
Executed_Gtid_Set: 73fdfd2a-9e36-11e5-8592-00a64151d129:1-138
1 row in set (0.00 sec)

而在从机上 SHOW SLAVE STATUS 你会看到一些错误：

Last_Error: Error ‘Can‘t create database ‘mysql‘; database exists‘ on query. Default database: ‘mysql‘. Query: ‘CREATE DATABASE mysql;

以及 RETRIEVED GTID SET  会显示 已取回138个记录复制到了从机：

mysql> SHOW SLAVE STATUS\\G
...
Retrieved_Gtid_Set: 73fdfd2a-9e36-11e5-8592-00a64151d129:1-138
..　　

当然，如果配置完成前关掉了GTID那就不会有这些错误了。

 

三台机器配置完成之后，输入 SHOW MASTER STATUS 显示如下：

mysql> SHOW MASTER STATUS\\G
*************************** 1. row ***************************
File: mysql-bin.000002
Position: 398
Binlog_Do_DB:
Binlog_Ignore_DB:
Executed_Gtid_Set:
1 row in set (0.00 sec)

 

到此，整个架构已经创建，下面进行一些测试。如图，创建一个漫画收藏数据库：

从机

CREATE DATABASE `comicbookdb`;
use comicbookdb;
CREATE TABLE `comics` (
`comic_id` int(9) NOT NULL AUTO_INCREMENT,
`comic_title` varchar(60) NOT NULL,
`issue_number` decimal(9,0) NOT NULL,
`pub_year` varchar(60) NOT NULL,
`pub_month` varchar(60) NOT NULL,
PRIMARY KEY (`comic_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1;　　

在主机和从机上都可以运行同样的SQL 语句。（因为在主机的一些列上我们用了AUTO_INCREMENT ，你可能觉得从机上就不用AUTO_INCREMENT 了。但是因为我们不会对从机做任何写入操作，所以你仍旧可以运行同样的语句，之后再去修改主键。后文会有详述。）

 

当数据从多个主机复制合并到一个从机时，复制程序将会处理 AUTO_INCREMENT 的问题。

下面在从机上创建表时对于comic_id 列时，没有声明 AUTO_INCREMENT出现了错误，

mysql> SHOW SLAVE STATUS\\G...
Last_SQL_Error: Error ‘Field ‘comic_id‘ doesn‘t have a default value‘ on query. Default database: ‘comicbookdb‘. Query: ‘INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘5‘,‘2014‘,‘03‘)‘
...

为了处理comic_id的主键问题，最方便的方法就是在主机中配置 auto_increment_increment ，在my.cnf  或者是my.ini 中：

[mysqld]
auto_increment_increment = 2　　

加这个变量需要重启mysql服务，但是你也可以直接在命令行操作，如下：

mysql> SET @@auto_increment_increment=2;
Query OK, 0 rows affected (0.00 sec)

验证一下：

mysql> SHOW VARIABLES WHERE VARIABLES_NAME = ‘auto_increment_increment‘;
+-----------------------------+-------+
| Variable_name | Value |
+-----------------------------+-------+
| auto_increment_increment | 2 |
+-----------------------------+-------+
1 row in set (0.00 sec)　　

如上 每个主键的增量就是2，只要每个主机的初始值不同就可以了。但不可以简单的设 0 或 1，因为如果是0的话，它会默认恢复到1，这样会引起冲突。所以我们设一个较大的值，最低位设置为 0 和 1，代码如下：

主机 # 1

CREATE DATABASE `comicbookdb`;
use comicbookdb;
CREATE TABLE `comics` (
`comic_id` int(9) NOT NULL AUTO_INCREMENT,
`comic_title` varchar(60) NOT NULL,
`issue_number` decimal(9,0) NOT NULL,
`pub_year` varchar(60) NOT NULL,
`pub_month` varchar(60) NOT NULL,
PRIMARY KEY (`comic_id`)
) ENGINE=InnoDB AUTO_INCREMENT=100000;

 

主机 # 2

CREATE DATABASE `comicbookdb`;
use comicbookdb;
CREATE TABLE `comics` (
`comic_id` int(9) NOT NULL AUTO_INCREMENT,
`comic_title` varchar(60) NOT NULL,
`issue_number` decimal(9,0) NOT NULL,
`pub_year` varchar(60) NOT NULL,
`pub_month` varchar(60) NOT NULL,
PRIMARY KEY (`comic_id`)
) ENGINE=InnoDB AUTO_INCREMENT=100001;

建表完毕，在所有主机上启动 GTID。这里从机也启动了GTID， 以防之后在这部从机之下加另一台从机。启动GTID，需要修改配置：

[mysqld]
auto_increment_increment = 2
gtid-mode = on
enforce-gtid-consistency = 1

重启每个server之后，可以检查一下每一台主机状态，状态一致：

mysql> SHOW MASTER STATUS\\G
*************************** 1. row ***************************
File: mysql-bin.000005
Position: 154
Binlog_Do_DB:
Binlog_Ignore_DB:
Executed_Gtid_Set:
1 row in set (0.00 sec)

接下来对所有的主从都做了一个服务器重置（reset master），非必要。重置会清空所有的binnary log ，并且新建一个二值日志。

mysql> RESET MASTER;
Query OK, 0 rows affected (0.00 sec)

mysql> SHOW MASTER STATUS\\G
*************************** 1. row ***************************
File: mysql-bin.000001
Position: 154
Binlog_Do_DB:
Binlog_Ignore_DB:
Executed_Gtid_Set:
1 row in set (0.00 sec)　

可以见到新的 binary log（mysql-bin.000001），开始位置是154.  我们往主机插入一些数据，然后再看看主机状态。

 

主机 #1

mysql> INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘1‘,‘2014‘,‘01‘);
Query OK, 1 row affected (0.02 sec)

mysql> SHOW MASTER STATUS\\G
*************************** 1. row ***************************
File: mysql-bin.000001
Position: 574
Binlog_Do_DB:
Binlog_Ignore_DB:
Executed_Gtid_Set: 63a7971c-b48c-11e5-87cf-f7b6a723ba3d:1
1 row in set (0.00 sec)

可见 插入的 GTID 是 63a7971c-b48c-11e5-87cf-f7b6a723ba3d:1 ，冒号前面部分是主机的UUID，这个信息可以在data目录下的auto.cnf里面查到。

主机 #1

# cat auto.cnf
[auto]
server-uuid=63a7971c-b48c-11e5-87cf-f7b6a723ba3d　　

再插入一行数据：

主机 #1

mysql> INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘2‘,‘2014‘,‘02‘);
Query OK, 1 row affected (0.05 sec)

mysql> show master status\\G
*************************** 1. row ***************************
File: mysql-bin.000001
Position: 994
Binlog_Do_DB:
Binlog_Ignore_DB:
Executed_Gtid_Set: 63a7971c-b48c-11e5-87cf-f7b6a723ba3d:1-2
1 row in set (0.00 sec)

mysql> select * from comics;
+----------+-------------+--------------+----------+-----------+
| comic_id | comic_title | issue_number | pub_year | pub_month |
+----------+-------------+--------------+----------+-----------+
| 100001 | Fly Man | 1 | 2014 | 01 |
| 100003 | Fly Man | 2 | 2014 | 02 |
+----------+-------------+--------------+----------+-----------+
2 rows in set (0.00 sec)　　

会发现这个数值变成了2，那么我们再在第二号主机插入两行数据，并且查看状态：

主机 #2

mysql> INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘3‘,‘2014‘,‘03‘);
mysql> INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘4‘,‘2014‘,‘04‘);

mysql> show master status\\G
*************************** 1. row ***************************
File: mysql-bin.000005
Position: 974
Binlog_Do_DB:
Binlog_Ignore_DB:
Executed_Gtid_Set: 75e2e1dc-b48e-11e5-83bb-1438deb0d51e:1-2
1 row in set (0.00 sec)

mysql> select * from comics;
+----------+-------------+--------------+----------+-----------+
| comic_id | comic_title | issue_number | pub_year | pub_month |
+----------+-------------+--------------+----------+-----------+
| 100002 | Fly Man | 3 | 2014 | 03 |
| 100004 | Fly Man | 4 | 2014 | 04 |
+----------+-------------+--------------+----------+-----------+
2 rows in set (0.00 sec)　　

主机#2有不同的UUID, 这也是我们分辨GTID对应于哪一步主机。那我们现在已经有两组GTID的会复制到从机上。当然，从机也有自己的UUID.

 

主机#1 跟主机#2的GTID设置：

63a7971c-b48c-11e5-87cf-f7b6a723ba3d:1-2
75e2e1dc-b48e-11e5-83bb-1438deb0d51e:1-2

　　

在此，通常我会确认下从机是没有运行的：

从机

mysql> show slave status\\G
Empty set (0.00 sec)　　

不同于平时的复制，在多主复制中，你需要为每一台主机创建一个通道，为这个通道命名。这里称之为 “master-142”（主机-142）和 “master-143”（主机143）去匹配server_id（就像IP一样）。接下来就演示如何开启主机#1(server_id=142)的数据复制。

 

从机

mysql> CHANGE MASTER TO MASTER_HOST=‘192.168.1.142‘, MASTER_USER=‘replicate‘, MASTER_PASSWORD=‘password‘, MASTER_AUTO_POSITION = 1 FOR CHANNEL ‘master-142‘;
Query OK, 0 rows affected, 2 warnings (0.23 sec)

这里有两个警告，可以忽略。

mysql> SHOW WARNINGS\\G
*************************** 1. row ***************************
Level: Note
Code: 1759
Message: Sending passwords in plain text without SSL/TLS is extremely insecure.
*************************** 2. row ***************************
Level: Note
Code: 1760
Message: Storing MySQL user name or password information in the master info repository is not secure and is therefore not recommended. Please consider using the USER and PASSWORD connection options for START SLAVE; see the ‘START SLAVE Syntax‘ in the MySQL Manual for more information.
2 rows in set (0.00 sec)　　

现在我们可以开启从机通道“master-142”：

mysql> START SLAVE FOR CHANNEL ‘master-142‘;
Query OK, 0 rows affected (0.03 sec)　　

这个命令同时启动了SQL_THREAD 跟 IO_THREAD。将来你会考虑停止一个或者多个线程，所以这里对应有一些语法，如何指定一个需要修改的通道：

START SLAVE SQL_THREAD FOR CHANNEL ‘master-142‘;
START SLAVE IO_THREAD FOR CHANNEL ‘master-142‘;　　

也可以发一个简单的命令“START SLAVE” 为需要执行复制操作的通道来开启这两个线程。从机启动，可以见到GTID被取出并且开始写入数据库。

mysql> SHOW SLAVE STATUS FOR CHANNEL ‘master-142‘\\G
*************************** 1. row ***************************
Slave_IO_State: Waiting for master to send event
Master_Host: 192.168.1.142
...
Master_UUID: 63a7971c-b48c-11e5-87cf-f7b6a723ba3d
...
Slave_SQL_Running_State: Slave has read all relay log; waiting for more updates
...
Retrieved_Gtid_Set: 63a7971c-b48c-11e5-87cf-f7b6a723ba3d:1-2
Executed_Gtid_Set: 63a7971c-b48c-11e5-87cf-f7b6a723ba3d:1-2
Auto_Position: 1
...
Channel_Name: master-142

 

检查从机，可以见到相关数据：　　

mysql> select * from comics;
+----------+-------------+--------------+----------+-----------+
| comic_id | comic_title | issue_number | pub_year | pub_month |
+----------+-------------+--------------+----------+-----------+
| 100001 | Fly Man | 1 | 2014 | 01 |
| 100003 | Fly Man | 2 | 2014 | 02 |
+----------+-------------+--------------+----------+-----------+
2 rows in set (0.00 sec)

 

主机#1 完成之后，我们开始配置主机#2：

CHANGE MASTER TO MASTER_HOST=‘192.168.1.143‘, MASTER_USER=‘replicate‘, MASTER_PASSWORD=‘password‘, MASTER_AUTO_POSITION = 1 FOR CHANNEL ‘master-143‘;　　

然后再次确认从机状态：

从机

mysql> SHOW SLAVE STATUS FOR CHANNEL ‘master-143‘\\G
*************************** 1. row ***************************
Slave_IO_State: Waiting for master to send event
Master_Host: 192.168.1.143
...
Master_UUID: 75e2e1dc-b48e-11e5-83bb-1438deb0d51e
...
Slave_SQL_Running_State: Slave has read all relay log; waiting for more updates
...
Retrieved_Gtid_Set: 75e2e1dc-b48e-11e5-83bb-1438deb0d51e:1-2
Executed_Gtid_Set: 63a7971c-b48c-11e5-87cf-f7b6a723ba3d:1-2,
75e2e1dc-b48e-11e5-83bb-1438deb0d51e:1-2,
Auto_Position: 1
...
Channel_Name: master-143　　

我们可以见到从机已经获取到两个GTID，并且已经在执行。再看看数据库表的内容，也能够发现四行数据都已经合并到了从机：

mysql> select * from comics;
+----------+-------------+--------------+----------+-----------+
| comic_id | comic_title | issue_number | pub_year | pub_month |
+----------+-------------+--------------+----------+-----------+
| 100001 | Fly Man | 1 | 2014 | 01 |
| 100002 | Fly Man | 3 | 2014 | 03 |
| 100003 | Fly Man | 2 | 2014 | 02 |
| 100004 | Fly Man | 4 | 2014 | 04 |
+----------+-------------+--------------+----------+-----------+
4 rows in set (0.01 sec)　　

复制过程处理了自增 AUTO_INCREMENT 的值。如果检查一下sql语句执行复制的原话，你会有所发现：

INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘1‘,‘2014‘,‘01‘)
INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘2‘,‘2014‘,‘02‘);
INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘3‘,‘2014‘,‘03‘);
INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘4‘,‘2014‘,‘04‘);　　

在主机上生成的自增值会随着声明传递给从机。检查主机上的binary log， 来到comic_id 这一列 “SET INSERT_ID = 100001”，整段会随着sql语句一起传给从机；

从机

# mysqlbinlog mysql-bin.000001
...
# at 349
#160106 21:08:01 server id 142 end_log_pos 349 CRC32 0x48fb16a2 Intvar
SET INSERT_ID=100001/*!*/;
#160106 21:08:01 server id 142 end_log_pos 543 CRC32 0xbaf55210 Query thread_id=1exec_time=0 error_code=0
use `comicbookdb`/*!*/;
SET TIMESTAMP=1452132481/*!*/;
INSERT INTO COMICS (comic_title, issue_number, pub_year, pub_month) VALUES(‘Fly Man‘,‘1‘,‘2014‘,‘01‘)
/*!*/;
...

 

整个讲解完毕，请大家多多指教。 

 

 

参考：

1，MySQL 5.7 Multi-Source Replication – Automatically Combining Data From Multiple Databases Into One

https://mysqlhighavailability.com/mysql-5-7-multi-source-replication-automatically-combining-data-from-multiple-databases-into-one/ 

2， MySQL Multi-Source Replication Overview

https://dev.mysql.com/doc/refman/5.7/en/replication-multi-source-overview.html