快捷導(dǎo)航

Python MySQL如何通過Binlog獲取變更記錄恢復(fù)數(shù)據(jù)

更新時(shí)間：2025年01月22日 14:50:47 作者：王小工

本文介紹了如何使用Python和pymysqlreplication庫通過MySQL的二進(jìn)制日志（Binlog）獲取數(shù)據(jù)庫的變更記錄,并展示了一個(gè)簡(jiǎn)單的Python腳本,該腳本讀取Binlog事件并打印出插入、更新和刪除操作的SQL語句,此外,還提到可以使用pandas將結(jié)果輸出到Excel表格中進(jìn)行數(shù)據(jù)分析處理

Python MySQL通過Binlog獲取變更記錄恢復(fù)數(shù)據(jù)

通過MySQL的二進(jìn)制日志（Binlog）獲取數(shù)據(jù)庫的變更記錄，并用于恢復(fù)數(shù)據(jù)，是一個(gè)相對(duì)高級(jí)的操作。

這通常涉及讀取Binlog中的事件，解析這些事件以了解數(shù)據(jù)變更的詳細(xì)信息，然后基于這些信息來恢復(fù)或回滾數(shù)據(jù)。

在Python中，你可以使用pymysqlreplication庫來讀取Binlog，但請(qǐng)注意，這個(gè)庫本身并不提供直接的數(shù)據(jù)恢復(fù)功能。它只能幫助你解析Binlog中的事件?；謴?fù)數(shù)據(jù)需要你根據(jù)這些事件編寫額外的邏輯。

以下是一個(gè)使用pymysqlreplication庫通過Binlog獲取MySQL操作記錄的示例：

1.安裝pymysqlreplication

首先，你需要安裝這個(gè)庫。你可以使用pip來安裝：

pip install pymysqlreplication

2.配置MySQL

確保你的MySQL服務(wù)器啟用了Binlog，并且你有一個(gè)具有足夠權(quán)限的MySQL用戶來讀取Binlog。

3.編寫Python腳本

下面是一個(gè)簡(jiǎn)單的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'
}
# 要監(jiān)控的數(shù)據(jù)庫和表 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服務(wù)器上的其他復(fù)制客戶端不同
        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()

在這個(gè)腳本中：

MYSQL_SETTINGS：包含了連接到MySQL服務(wù)器所需的設(shè)置。
BinLogStreamReader：包含了讀取Binlog所需的設(shè)置，包括server_id（必須是一個(gè)唯一的標(biāo)識(shí)符，用于區(qū)分不同的復(fù)制客戶端）和only_events（指定我們感興趣的事件類型）。
stream：函數(shù)根據(jù)事件的類型（刪除、更新或插入）打印出相應(yīng)的SQL語句。
main：函數(shù)設(shè)置了Binlog流讀取器，并在捕獲到任何異常時(shí)優(yōu)雅地關(guān)閉流。
pandas：將結(jié)果輸出到excel表格中，用于數(shù)據(jù)進(jìn)行分析處理。