最近，我一直在与一些客户合作，他们的MySQL集群上有多TB的事务数据。这些非常大的数据集并不是他们日常操作所真正需要的，但它们非常方便，因为它们允许他们轻松地查询历史数据。然而，这种便利的代价很高，您要为存储付出更多的代价，备份和恢复所需的时间要长得多，而且当然要大得多。因此，问题是：他们如何执行“高效数据归档”？

让我们尝试定义什么是高效的数据归档体系结构。我们可以布置一些关键要求：

• 存档应位于异步副本上

• 存档副本应使用针对大型数据集优化的存储配置

• 常规集群应该只是正常删除数据

• 存档系统应从复制流中删除删除语句，并仅保留插入和更新

• 归档系统应该是健壮的，能够处理故障并恢复复制

关键要素

我们的初始起点如下：

集群由一个源和两个副本（R1和R2）组成，我们正在为存档添加一个副本（RA）。只要基于行的复制格式与全行映像一起使用，现有集群在接下来的所有讨论中都几乎不相关。

理论上，上述设置足以存档数据，但为了做到这一点，我们不能允许要存档的表上的delete语句流经复制流。必须在所有正常服务器上使用sql_log_bin=0执行删除。虽然这看起来很简单，但它有许多缺点。必须在所有服务器上定期调用cron作业或SQL事件。这些作业必须删除所有生产服务器上的相同数据。这个过程可能会在表之间引入一些差异。pt表校验和等验证工具可能开始报告误报。我们将看到，还有其他选择。

捕捉变化（CDC）

我们需要的一个重要组件是捕获要归档的表的更改的方法。当使用基于行的格式和全行图像时，MySQL二进制日志非常适合这个目的。我们需要一个工具，它可以连接到数据库服务器（如副本），将二进制日志事件转换为可用形式，并跟踪其在二进制日志中的位置。

对于这个项目，我们将使用由Zendesk开发的工具Maxwell。Maxwell连接到源服务器，就像常规副本一样，并以JSON格式输出基于行的事件。它在源服务器上的表中跟踪其复制位置。

移除删除

由于CDC组件将以JSON格式输出事件，我们只需要过滤我们感兴趣的表，然后忽略删除事件。您可以使用任何支持JSON和MySQL的编程语言。在本文中，我将使用Python。

档案存储引擎

InnoDB对于事务性工作负载非常有用，但对于归档数据则远远不够理想。MyRocks是一个更好的选择，因为它是写优化的，在数据压缩方面效率更高。

用于高效数据归档的体系结构

移动表

我们为存档副本提供了一些体系结构选项。第一种体系结构（如下所示）将CDC连接到存档副本。这意味着，如果要存档表t，则需要在存档副本上同时保留生产t（从中删除数据）和存档副本tA（长期保存数据）。

此体系结构的主要优点是，与归档过程相关的所有组件都只与归档副本交互。当然，消极的一面是存档副本上存在重复数据，因为它必须同时承载t和tA。有人可能会说，t表可能使用了黑洞存储引擎，但我们不要潜到这样一个兔子洞里。

忽略表

另一个体系结构选项是使用来自源的两个不同的复制流。第一个流是常规复制链接，但副本具有复制选项replicate ignore table=t。表t的复制事件由Maxwell控制的第二复制链路处理。删除事件将被删除，插入和更新将应用于存档副本。

虽然稍后的体系结构仅在存档副本上存储t的单个副本，但它需要来自源的两个完整复制流。

实例

应用程序

我现在的目标是提供一个尽可能简单的示例，同时仍然有效。我将在Sysbench tpc-c脚本中使用移位表方法。该脚本有一个选项enable_purge，用于删除已处理的旧订单。我们的目标是创建表tpccArchive。orders1包含所有行，甚至是删除的行，而表tpcc。orders1是常规订单表。它们具有相同的结构，但存档表使用MyRocks。

让我们首先准备归档表：

mysql> create database tpccArchive;
Query OK, 1 row affected (0,01 sec)
 
mysql> use tpccArchive;
Database changed
 
mysql> create table orders1 like tpcc.orders1;
Query OK, 0 rows affected (0,05 sec)
 
mysql> alter table orders1 engine=rocksdb;
Query OK, 0 rows affected (0,07 sec)
Records: 0  Duplicates: 0  Warnings: 0

捕捉变化

现在，我们可以安装麦克斯韦了。Maxwell是一个基于Java的应用程序，因此需要一个兼容的JRE。它还将作为副本连接到MySQL，因此它需要一个具有所需授权的帐户。它还需要自己的maxwell模式，以保持复制状态和位置。

root@LabPS8_1:~# apt-get install openjdk-17-jre-headless 
root@LabPS8_1:~# mysql -e "create user maxwell@'localhost' identified by 'maxwell';"
root@LabPS8_1:~# mysql -e 'create database maxwell;'
root@LabPS8_1:~# mysql -e 'grant ALL PRIVILEGES ON maxwell.* TO maxwell@localhost;'
root@LabPS8_1:~# mysql -e 'grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO maxwell@localhost;'
root@LabPS8_1:~# curl -sLo - https://github.com/zendesk/maxwell/releases/download/v1.37.6/maxwell-1.37.6.tar.gz| tar zxvf -
root@LabPS8_1:~# cd maxwell-1.37.6/
root@LabPS8_1:~/maxwell-1.37.6# ./bin/maxwell -help
Help for Maxwell:

Option                   Description                                                                       
------                   -----------                                                                       
--config <String>        location of config.properties file                                                
--env_config <String>    json object encoded config in an environment variable                             
--producer <String>      producer type: stdout|file|kafka|kinesis|nats|pubsub|sns|sqs|rabbitmq|redis|custom
--client_id <String>     unique identifier for this maxwell instance, use when running multiple maxwells   
--host <String>          main mysql host (contains `maxwell` database)                                     
--port <Integer>         port for host                                                                     
--user <String>          username for host                                                                 
--password <String>      password for host                                                                 
--help [ all, mysql, operation, custom_producer, file_producer, kafka, kinesis, sqs, sns, nats, pubsub, output, filtering, rabbitmq, redis, metrics, http ]


In our example, we’ll use the <i>stdout</i> producer to keep things as simple as possible. 

过滤脚本

以便向tpccArchive添加和更新行。对于orders1表，我们需要一段逻辑来标识表tpcc的事件。orders1并忽略delete语句。同样，为了简单起见，我选择使用Python脚本。我不会在这里展示整个脚本，欢迎从我的GitHub存储库下载。它本质上是一个在线循环，写入到stdin。该行作为JSON字符串加载，然后根据找到的值做出一些决定。下面是其核心的一小部分代码：

...
for line in sys.stdin:
    j = json.loads(line)
    if j['database'] == dbName and j['table'] == tableName:
        debug_print(line)
        if j['type'] == 'insert':
            # Let's build an insert ignore statement
             sql += 'insert ignore into ' + destDbName + '.' + tableName
...

当事件类型为“insert”时，上一节将创建“INSERTEIGNORE”语句。脚本使用用户归档程序和密码tpcc连接到数据库，然后将事件应用到表tpccArchive.orders1。

root@LabPS8_1:~# mysql -e "create user archiver@'localhost' identified by 'tpcc';"
root@LabPS8_1:~# mysql -e 'grant ALL PRIVILEGES ON tpccArchive

为了便于复制这些步骤，下面是应用程序（tpcc）方面：

yves@ThinkPad-P51:~/src/sysbench-tpcc$ ./tpcc.lua --mysql-host=10.0.4.158 --mysql-user=tpcc --mysql-password=tpcc --mysql-db=tpcc \
        --threads=1 --tables=1 --scale=1 --db-driver=mysql --enable_purge=yes --time=7200 --report-interval=10 prepare
yves@ThinkPad-P51:~/src/sysbench-tpcc$ ./tpcc.lua --mysql-host=10.0.4.158 --mysql-user=tpcc --mysql-password=tpcc --mysql-db=tpcc \
        --threads=1 --tables=1 --scale=1 --db-driver=mysql --enable_purge=yes --time=7200 --report-interval=10 run

数据库正在运行一个IP为10.0.4.158的虚拟机。enable_purge选项会导致旧订单1被删除。对于存档端，在数据库VM上运行：

root@LabPS8_1:~/maxwell-1.37.6# bin/maxwell --user='maxwell' --password='maxwell' --host='127.0.0.1' \
        --producer=stdout 2> /tmp/maxerr | python3 ArchiveTpccOrders1.py

在两个小时的tpcc运行后，我们有：

mysql> select  TABLE_SCHEMA, TABLE_ROWS, DATA_LENGTH, INDEX_LENGTH, ENGINE from information_schema.tables where table_name='orders1';
+--------------+------------+-------------+--------------+---------+
| TABLE_SCHEMA | TABLE_ROWS | DATA_LENGTH | INDEX_LENGTH | ENGINE  |
+--------------+------------+-------------+--------------+---------+
| tpcc         |      48724 |     4210688 |      2310144 | InnoDB  |
| tpccArchive  |    1858878 |    38107132 |     14870912 | ROCKSDB |
+--------------+------------+-------------+--------------+---------+
2 rows in set (0,00 sec)

上面的例子就是一个例子。任何生产系统都需要比我的示例更加坚固。以下是一些要求：

• Maxwell必须能够从正确的复制位置重新启动并继续

• Python脚本必须能够从正确的复制位置重新启动并继续

• 如果连接断开，Python脚本必须能够重新连接到MySQL并重试事务。

麦克斯韦已经处理了第一点，它使用数据库来存储其当前位置。

下面的逻辑步骤是在Maxwell和Python脚本之间添加一个比简单的进程管道更健壮的队列系统。Maxwell支持许多排队系统，如kafka、kinesis、rabbitmq、redis等。对于我们的应用程序，我倾向于使用kafka和单个分区的解决方案。kafka不管理消息的偏移量，这取决于应用程序。这意味着Python脚本可以更新表中的一行，作为其应用的每个事务的一部分，以跟踪其在kafka流中的位置。如果存档表使用RocksDB，则队列位置跟踪表也应使用RocksDB，以便数据库事务不跨存储引擎。

结论

在本文中，我提供了一个使用MySQL复制二进制日志归档数据的解决方案。归档快速增长的表是经常需要的，希望这样的解决方案能有所帮助。如果在副本上有一个MySQL插件能够直接过滤复制事件，那就太好了。这将不再需要像Maxwell和我的python脚本这样的外部解决方案。但是，一般来说，这种归档解决方案只是汇总表的一种特殊情况。在未来的帖子中，我希望能提供一个更完整的解决方案，并保留一个摘要。

原文标题：Efficient Data Archiving in MySQL
原文作者：Yves Trudeau
原文链接：https://www.percona.com/blog/efficient-data-archiving-in-mysql/