首页 > 脚本专栏 > python > Python MySQL通过Binlog获取变更记录恢复数据

Python MySQL如何通过Binlog获取变更记录恢复数据

2025-01-22 14:50:47 作者：王小工

本文介绍了如何使用Python和pymysqlreplication库通过MySQL的二进制日志（Binlog）获取数据库的变更记录,并展示了一个简单的Python脚本,该脚本读取Binlog事件并打印出插入、更新和删除操作的SQL语句,此外,还提到可以使用pandas将结果输出到Excel表格中进行数据分析处理

Python MySQL通过Binlog获取变更记录恢复数据

通过MySQL的二进制日志（Binlog）获取数据库的变更记录，并用于恢复数据，是一个相对高级的操作。

这通常涉及读取Binlog中的事件，解析这些事件以了解数据变更的详细信息，然后基于这些信息来恢复或回滚数据。

在Python中，你可以使用pymysqlreplication库来读取Binlog，但请注意，这个库本身并不提供直接的数据恢复功能。它只能帮助你解析Binlog中的事件。恢复数据需要你根据这些事件编写额外的逻辑。

以下是一个使用pymysqlreplication库通过Binlog获取MySQL操作记录的示例：

1.安装pymysqlreplication

首先，你需要安装这个库。你可以使用pip来安装：

pip install pymysqlreplication

2.配置MySQL

确保你的MySQL服务器启用了Binlog，并且你有一个具有足够权限的MySQL用户来读取Binlog。

3.编写Python脚本

下面是一个简单的Python脚本，它使用pymysqlreplication.BinLogStreamReader来读取Binlog事件，并打印出插入、更新和删除操作的信息。

import json
import sys
from datetime import datetime
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
    DeleteRowsEvent,
    UpdateRowsEvent,
    WriteRowsEvent,
)
import pandas as pd

MYSQL_SETTINGS = {
    'host': '127.0.0.1',
    'port': 3306,
    'user': 'root',
    'password': '123456'
}
# 要监控的数据库和表 ssc_sjzz2
database_name = 'ssc_wfg'
table_name = 't_sys_user'

def default(o):
    if isinstance(o, datetime):
        return o.isoformat()
    raise TypeError("Unserializable object {}".format(o))
def main():
    stream = BinLogStreamReader(
        connection_settings=MYSQL_SETTINGS,
        server_id=6, # 必须与MySQL服务器上的其他复制客户端不同
        only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent],
        only_tables=[table_name],
        only_schemas=[database_name]
        ) 
    df = pd.DataFrame()
    for binlogevent in stream:
        if binlogevent.table == table_name and binlogevent.schema == database_name:
            time = binlogevent.formatted_timestamp.replace('T', ' ')
            timestamp= binlogevent.timestamp
            for row in binlogevent.rows:
                event = {
                    "schema": binlogevent.schema, 
                    "table": binlogevent.table,
                    "time": time,
                    "timestamp": timestamp,
                    }
            
                if isinstance(binlogevent, DeleteRowsEvent):
                    event["action"] = "delete"
                    event["value"] = json.dumps(list(row["values"].items()), default=default)
                    # event = dict(event.items() + row["values"].items())
                elif isinstance(binlogevent, UpdateRowsEvent):
                    event["action"] = "update"
                    event["value"] = json.dumps(list(row["after_values"].items()), default=default)
                    # event = dict(event.items() + row["after_values"].items())
                elif isinstance(binlogevent, WriteRowsEvent):
                    event["action"] = "insert"
                    event["value"] = json.dumps(list(row["values"].items()), default=default)
                print(json.dumps(event, default=default))
                df = pd.concat([df, pd.DataFrame(event, index=[0])], ignore_index=True)
    stream.close()
    df.to_excel('binlog.xlsx', index=False)
if __name__ == "__main__":
    main()

在这个脚本中：

MYSQL_SETTINGS：包含了连接到MySQL服务器所需的设置。
BinLogStreamReader：包含了读取Binlog所需的设置，包括server_id（必须是一个唯一的标识符，用于区分不同的复制客户端）和only_events（指定我们感兴趣的事件类型）。
stream：函数根据事件的类型（删除、更新或插入）打印出相应的SQL语句。
main：函数设置了Binlog流读取器，并在捕获到任何异常时优雅地关闭流。
pandas：将结果输出到excel表格中，用于数据进行分析处理。

4.运行脚本

运行这个Python脚本，它将连接到你的MySQL服务器，并开始读取Binlog中的事件。

每当有新的事件发生时（如插入、更新或删除操作），它都会打印出相应的SQL语句。

总结

以上为个人经验，希望能给大家一个参考，也希望大家多多支持脚本之家。