pt-archiver mysql数据归档

1、说明

pt-archiver 是 percona-toolkit 高级命令行工具集中的一种，可以查看当前服务的摘要信息，磁盘检测，分析慢查询日志，查找重复索引，实现表同步等等；

2、 安装

这里以centos7为例进行安装说明

yum -y install  perl-DBI  perl-DBD-MySQL  perl-TermReadKey perl-devel perl-Time-HiRes
cd /usr/local/src/
wget https://www.percona.com/downloads/percona-toolkit/2.2.15/deb/percona-toolkit_2.2.15-2.tar.gz
tar -zxf percona-toolkit_2.2.15-2.tar.gz
cd percona-toolkit-2.2.15/
chmod +x Makefile.PL
perl Makefile.PL
make && make install
ln -s /usr/bin  /usr/local/bin/*

3、语法

pt-archiver [OPTIONS] --source DSN --where WHERE

参数说明：

--limit10000 每次取1000行数据用pt-archive处理，Number of rows to fetch and archive per statement.
--txn-size 1000 设置1000行为一个事务提交一次，Number of rows pertransaction.
--where'id<3000' 设置操作条件
--progress5000 每处理5000行输出一次处理信息
--statistics 输出执行过程及最后的操作统计。（只要不加上--quiet，默认情况下pt-archive都会输出执行过程的）
--charset=UTF8 指定字符集为UTF8
--bulk-delete 批量删除source上的旧数据(例如每次1000行的批量删除操作)
--bulk-insert 批量插入数据到dest主机 (看dest的general log发现它是通过在dest主机上LOAD DATA LOCAL INFILE插入数据的)
--replace 将insert into 语句改成replace写入到dest库
--sleep120 每次归档了limit个行记录后的休眠120秒（单位为秒）
--file'/root/test.txt'
--purge 删除source数据库的相关匹配记录
--header 输入列名称到首行（和--file一起使用）
--no-check-charset 不指定字符集
--check-columns 检验dest和source的表结构是否一致，不一致自动拒绝执行（不加这个参数也行。默认就是执行检查的）
--no-check-columns 不检验dest和source的表结构是否一致，不一致也执行（会导致dest上的无法与source匹配的列值被置为null或者0）
--chekc-interval 默认1s检查一次
--local 不把optimize或analyze操作写入到binlog里面（防止造成主从延迟巨大）
--retries 超时或者出现死锁的话，pt-archiver进行重试的间隔（默认1s）
--no-version-check 目前为止，发现部分pt工具对阿里云RDS操作必须加这个参数
--analyze=ds 操作结束后，优化表空间（d表示dest，s表示source）

4、归档到数据库

复制数据到其他mysql实例，并删source上的旧数据(指定字符集)：
pt-archiver \
--source h=localhost,u=archiver,p=archiver,P=3306,D=test,t=t1 \
--dest h=192.168.2.12,P=3306,u=archiver,p=archiver,D=test,t=t1_his \
--progress 5000 --where "c <'2017-05-01 00:00:00' " \
--statistics --charset=UTF8 --limit=10000 --txn-size 1000 --no-delete  --bulk-insert

5、归档到文件

pt-archiver \
--source h=10.0.20.26,u=root,p=1234,P=3306,D=test,t=t \
--file '/root/test.txt' \
--progress 5000 --where 'a<12000' \
--statistics --charset=UTF8 --limit=10000 --txn-size 1000 --purge

6、结合python归档

在Python中提供了很多的方法可以调用并执行shell脚本

• os.system(“command”) python自带的执行shell命令的方法，其中最后一个0是这个命令的返回值，为0表示命令执行成功。但是使用system()无法将执行的结果保存起来。
• os.popen() 返回的是 file read 的对象，对其进行读取 read() 的操作可以看到执行的输出。

这里以同库不同表归档为例，可以根据需要自行拓展配置
config.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-

class MysqlConfig:
    host = ""
    port = 3306
    username = "root"
    password = ""
    schema = "db-create"
    table = ""
    bakTable = ""
    condition = "date < '2022-04-01'"

main.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import os
import config
import sys
import time
import logging

if __name__ == '__main__':
    # 设置归档时间
    start = time.strftime('%Y-%m-%d %H:%M:%S')
    # 设置日志
    LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
    logging.basicConfig(level=logging.DEBUG,format=LOG_FORMAT,filename='log.log')
    logging.info("归档开始："+start)

    server_source = config.MysqlConfig.host
    port_source = config.MysqlConfig.port
    user_source = config.MysqlConfig.username
    password_source = config.MysqlConfig.password
    db_source = config.MysqlConfig.schema
    table_source = config.MysqlConfig.table
    bak_source = config.MysqlConfig.bakTable
    archive_condition = config.MysqlConfig.condition

    # pt-archive 命令
    archive_cmd = "pt-archiver "\
                  "--source h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' "\
                  "--dest h='%s',P='%s',u='%s',p='%s',D='%s',t='%s' "\
                  "--charset=UTF8 --where \"%s\" --progress 50000 --limit 10000 --txn-size 10000 "\
                  "--bulk-insert --bulk-delete --statistics --no-check-charset --purge" % \
                  (server_source, port_source, user_source, password_source, db_source, table_source,
                   server_source, port_source, user_source, password_source, db_source, bak_source,
                   archive_condition)
    # 获取系统输出
    stdout_archive = sys.stdout
    log_file = open('/home/python_script/db_archive_%s_%s.log' % (db_source, table_source), "w")
    # redirect print output to log file
    sys.stdout = log_file
    # archive_cmd = os.popen(pt_archive)
    with os.popen(archive_cmd) as c:
        # with open("db_archive1.log", "r") as c:
        archive_log = c.read()
        print(archive_cmd)
        print(archive_log)

    # close log file
    log_file.close()
    # restore the output to initial pattern
    sys.stdout = stdout_archive

    # 归档结束时间
    end = time.strftime('%Y-%m-%d %H:%M:%S')
    print("执行完成")
    logging.info("归档结束："+end)

7、参考资料

https://cloud.tencent.com/developer/article/1508825
https://www.modb.pro/db/329686

提供一个测试 数据库
https://github.com/wenxpro/db_archiver/tree/main/db
-end-