## 背景

部署在线上服务器的MySQL数据库中有很多用来临时使用的数据库，想着清理一下，但删库一时爽，然后没刹住车，把线上一个小程序的在用唯一的数据库删掉了，并且这个数据库我没有任何备份机制。
当前案例中的服务器操作系统为Ubuntu，MySQL数据库版本为`8.0`，且部署在Docker容器中，容器名称为`mysql`，下文均以此条件开展。

## 是否具备恢复条件

### 检查是否已开启Binlog

PS：官方版本的MySQL是默认开启binlog的。
检查是否开启了binlog（二进制日志），将下面命令中的容器名称替换成你实际的容器名：
```
docker exec -it <容器名称> mysql -u root -p
```
执行后会让你输入你MySQL中root账户密码，密码正确后会进入MySQL命令行，在MySQL命令行中输入以下命令查看是否启用了binlog：
```
SHOW VARIABLES LIKE 'log_bin';
```
输出结果为：
```
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| log_bin       | ON    |
+---------------+-------+
```
表示binlog已开启，一般是默认开启的。
### 检查Binlog格式

进一步确认binlog格式，最好是ROW 或 MIXED，依然在MySQL命令行中执行：
```
SHOW VARIABLES LIKE 'binlog_format';
```
结果输出为：
```
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW   |
+---------------+-------+
```
这是最理想的状态，基本上MySQL的任何操作都会被记录下来。
### 检查是否存在已记录的Binlog文件

继续在MySQL命令行中执行命令查看当前MySQL有哪些binlog文件：
```
SHOW BINARY LOGS;
```
结果输出为：
```
+---------------+-----------+-----------+
| Log_name      | File_size | Encrypted |
+---------------+-----------+-----------+
| binlog.000002 |  19625064 | No        |
| binlog.000003 |    853437 | No        |
+---------------+-----------+-----------+
```
该输出表示有两个binlog文件，并且没有进行加密。到此为止基本上一切工作都准备就绪，基本具备恢复条件。

## 恢复操作

### 复制日志文件到宿主机

将docker容器中的日志文件复制到宿主机中：
```
docker cp <容器名称>:/var/lib/mysql/binlog.000002 ./recovery/
docker cp <容器名称>:/var/lib/mysql/binlog.000003 ./recovery/
```
其中`recovery`为宿主机的目录，根据你实际的目录路径填写。
### 安装日志解析工具包

利用本地的MySQL服务工具包中的mysqlbinlog根据对日志进行解析，先在本地执行一下以下命令，看一下本地是否有相关服务组件：
```
mysqlbinlog --version
```
如果能够正常输出版本号则表示有相关服务，如果没有需要通过下面命令执行安装：
```
apt update
apt install mysql-server-core-8.0 -y
```
安装完成后继续执行上面查看mysqlbinlog版本号的操作，确认一下是否安装成功。
### 解析日志文件

mysqlbinlog命令安装成功后使用如下命令对刚刚从容器中复制出来的日志文件进行解析并将其输出到可读文件中：
```
mysqlbinlog \
  --base64-output=DECODE-ROWS \
  -v \
  binlog.000002 \
  binlog.000003 > full_binlog.sql
```
如上命令会将两个日志文件按照时间顺序统一输出到full_binlog.sql文件中。
到这里，最好的方式是自己打开这个文件，寻找要恢复的那个数据库的建库语句，比如如下所示：
```
CREATE DATABASE `tworice_bill` CHARACTER SET 'utf8mb4'
/*!*/;
```
手动将此条数据之前的内容删掉，此步骤是为了防止日志文件过大导致后续解析脚本没必要的算力，也为了方便提取建表语句。
### 提取建表语句

现在我们就得到了从目标数据库建库开始的完整日志文件了，接下来要提取其中与该数据库相关的`CREATE TABLE`、`INSERT`、`UPDATE`、`DELETE`命令。
将建表语句和数据行操作语句的提取操作分开，也就是先将`CREATE TABLE`相关命令提取出来到一个文件，因为篇幅原因这里就不贴出完整的脚本代码了，放到公众号后台获取。
拿到脚本代码后将get_table.py脚本文件中目标表名称调整为你实际的表名称，如果你多个数据库有相同的数据表也不要怕，这个脚本文件会提取第一次出现该建表语句的命令，所以这就是为什么上面要手动将建库语句之前的内容全部删掉的原因。
![image](https://developer.qcloudimg.com/http-save/yehe-9738124/242d825fdfbe4fd840dab69b0216795d.png)
将上图中表名称调整完成后记得调整一下脚本文件中`full_binlog.sql`文件绝对地址，之后就可以直接执行该脚本文件。执行结束后会输出`create_table_sql.sql`文件，这就是你要恢复的数据库所有建表语句。

### 提取数据行操作语句

依然是从公众号获取Binlog日志恢复脚本`get_insert_update.py`，调整脚本中`INPUT_FILE`、`OUTPUT_FILE `、`TARGET_DB`、`TABLE_FIELDS`之后执行该脚本，执行成功后会输出一个`insert_update_sql.sql`文件，该文件中会包含INSERT、UPDATE、DELETE语句，到数据库中依次执行上面输出的建表语句和本次输出的数据行操作语句即可完整恢复该数据库了。

## 最终效果

### 建表语句提取

建表语句在Binlog日志中比较好提取，基本是先找到`CREATE TABLE`字段后再找下一个`;`即可提取出来。
![image](https://developer.qcloudimg.com/http-save/yehe-9738124/8573ee7c1822e666209929cee8298f9e.png)
Binlog文件中原始建表日志

![image](https://developer.qcloudimg.com/http-save/yehe-9738124/72b4b18e9114fcf03cf7a029487504f1.png)
脚本提取恢复后的建表语句

### 数据行操作语句提取

Binlog记录的数据行操作日志是以字段索引记录的，而可执行SQL语句中像UPDATE需要精确字段名才可以。
![image](https://developer.qcloudimg.com/http-save/yehe-9738124/1ef8ce6b0c71003ed239217c2775334d.png)
Binlog文件中原始数据行操作日志
![image](https://developer.qcloudimg.com/http-save/yehe-9738124/49b0521605e80d9eace3e9b9e4b5d460.png)
脚本提取恢复后的数据行操作语句。

## 附完整脚本文件内容
解析指定数据库的建表语句脚本：
```
import re
import os

# 目标表集合（小写）
TARGET_TABLES = {
    'system_department',
    'system_deploy_log',
    'system_dict',
    'system_dict_value',
    'system_feedback',
    'system_feedback_type',
    'system_file',
    'system_login_log',
    'system_notice',
    'system_operation_log',
    'system_resources',
    'system_role',
    'system_role_resources',
    'system_user',
    'system_user_department',
    'system_user_info',
    'system_user_role',
    'system_white_list',
    'bill',
    'bill_classify',
    'bill_plan'
}

def extract_create_table_statements(input_path, output_path):
    extracted_tables = set()
    output_lines = []

    # 用于从 CREATE TABLE 行中提取表名的正则（支持反引号）
    table_name_pattern = re.compile(
        r'CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?`?(\w+)`?',
        re.IGNORECASE
    )

    with open(input_path, 'r', encoding='utf-8', errors='ignore') as f:
        capturing = False      # 是否正在捕获一个建表语句
        current_stmt_lines = []  # 当前正在捕获的语句行列表
        current_table_name = None  # 当前捕获的表名（小写）

        for line in f:
            stripped_line = line.rstrip()  # 去掉行尾换行符，保留内容

            if not capturing:
                # 检查当前行是否包含 CREATE TABLE
                if re.search(r'CREATE\s+TABLE', stripped_line, re.IGNORECASE):
                    # 尝试提取表名
                    match = table_name_pattern.search(stripped_line)
                    if match:
                        table_name = match.group(1).lower()
                        if table_name in TARGET_TABLES and table_name not in extracted_tables:
                            # 开始捕获
                            capturing = True
                            current_stmt_lines = [stripped_line]
                            current_table_name = table_name
                        # 否则：不是目标表，或已提取过，跳过
                    # else: 无法解析表名，跳过（如格式异常）
                # else: 普通行，跳过
            else:
                # 正在捕获中，追加当前行
                current_stmt_lines.append(stripped_line)

                # 判断是否结束：如果这一行包含独立的分号（通常在结尾）
                # 注意：MySQL DDL 中分号通常在行尾，且前面没有未闭合的括号或字符串（我们简化处理）
                if stripped_line.rstrip().endswith(';'):
                    # 结束捕获
                    full_stmt = '\n'.join(current_stmt_lines)
                    extracted_tables.add(current_table_name)
                    output_lines.append(full_stmt)
                    # 重置状态
                    capturing = False
                    current_stmt_lines = []
                    current_table_name = None

    # 写入输出文件
    with open(output_path, 'w', encoding='utf-8') as out_f:
        for stmt in output_lines:
            out_f.write(stmt + '\n\n')

    print(f"成功提取 {len(output_lines)} 个目标表的建表语句。")
    print(f"结果已保存至: {output_path}")

if __name__ == '__main__':
    input_file = r"E:\Study\script\recover\full_binlog.sql"
    output_file = "create_table_sql.sql"

    if not os.path.isfile(input_file):
        print("输入文件不存在，请检查路径：", input_file)
    else:
        extract_create_table_statements(input_file, output_file)
```
解析数据行操作语句脚本：
```
import re
import os
import sys

INPUT_FILE = r"E:\Study\script\recover\full_binlog.sql"
OUTPUT_FILE = r"E:\Study\script\recover\insert_update_sql.sql"

TARGET_DB = "tworice_bill"


# ===================== 表字段映射（必须补全） =====================
TABLE_FIELDS = {
    'bill': [
        'id', 'create_time', 'update_time', 'creator', 'classify',
        'type', 'price', 'remark', 'deleted', 'consume_time'
    ],

    'bill_classify': [
        'id', 'create_time', 'update_time', 'creator',
        'name', 'icon', 'deleted','open'
    ],

    'bill_plan': [
        'id', 'create_time', 'update_time', 'creator',
        'classify', 'spend_target', 'begin_date', 'end_date', 'deleted'
    ],

    'system_department': [
        'id', 'department_name', 'deleted'
    ],

    'system_deploy_log': [
        'id', 'create_time', 'update_time', 'creator',
        'state', 'user_id', 'deleted', 'deploy_id'
    ],

    'system_dict': [
        'id', 'create_time', 'name', 'description',
        'logo', 'update_time', 'creator', 'deleted'
    ],

    'system_dict_value': [
        'id', 'create_time', 'dict', 'content', 'description'
    ],

    'system_feedback': [
        'id', 'create_time', 'update_time', 'creator',
        'fb_type', 'fb_describe', 'fb_deal',
        'fb_state', 'fb_contact', 'deleted'
    ],

    'system_feedback_type': [
        'id', 'create_time', 'update_time', 'creator',
        'type_name', 'type_state', 'type_desc',
        'type_level', 'deleted'
    ],

    'system_file': [
        'id', 'create_time', 'update_time', 'path', 'name',
        'creator', 'deleted', 'type', 'parent', 'size',
        'share', 'level', 'pwd', 'state'
    ],

    'system_login_log': [
        'id', 'create_time', 'login_name',
        'pass_word', 'ip_addr', 'status', 'active'
    ],

    'system_notice': [
        'id', 'create_time', 'title', 'content',
        'top', 'creator', 'update_time', 'deleted'
    ],

    'system_operation_log': [
        'id', 'create_time', 'active', 'user_id',
        'table_name', 'target_id', 'ip_addr'
    ],

    'system_resources': [
        'id', 'name', 'url', 'pid', 'sort',
        'icon', 'type', 'component',
        'state', 'path', 'menu'
    ],

    'system_role': [
        'id', 'role_name', 'deleted',
        'create_time', 'update_time', 'creator'
    ],

    'system_role_resources': [
        'id', 'role_id', 'resource_id'
    ],

    'system_user': [
        'id', 'login_name', 'pass_word', 'nick_name',
        'status', 'create_time', 'update_time',
        'deleted', 'creator', 'invite_code', 'inviter_id',"email","phone","mp_openid"
    ],

    'system_user_department': [
        'id', 'department_id', 'user_id', 'deleted'
    ],

    'system_user_info': [
        'id', 'create_time', 'admin_sex', 'admin_phone',
        'admin_email', 'home_place', 'contact_address',
        'admin_note', 'user_id', 'update_time',
        'creator', 'deleted'
    ],

    'system_user_role': [
        'id', 'role_id', 'user_id', 'deleted'
    ],

    'system_white_list': [
        'id', 'create_time', 'update_time',
        'begin_ip', 'end_ip', 'state', 'deleted'
    ]
}
# ===================== 正则 =====================
RE_INSERT = re.compile(r"### INSERT INTO (.+)")
RE_UPDATE = re.compile(r"### UPDATE (.+)")
RE_DELETE = re.compile(r"### DELETE FROM (.+)")
RE_TABLE = re.compile(r"`?([^`.]+)`?\.`?([^`]+)`?")
RE_SET = re.compile(r"### SET")
RE_WHERE = re.compile(r"### WHERE")
RE_COL = re.compile(r"###\s+@(\d+)=(.+)")
RE_EVENT_END = re.compile(r"^# at |^COMMIT")

# ===================== 工具 =====================
def parse_table(expr):
    m = RE_TABLE.search(expr)
    if not m:
        return None, None
    return m.group(1), m.group(2)


def get_field(table, idx):
    fields = TABLE_FIELDS.get(table)
    if not fields or idx - 1 >= len(fields):
        raise IndexError(f"{table} @ {idx}")
    return fields[idx - 1]


def write_sql(f, sql):
    f.write(sql + ";\n")


# ===================== 主逻辑 =====================
def main():
    if not os.path.exists(INPUT_FILE):
        print("binlog file not found")
        sys.exit(1)

    with open(INPUT_FILE, "r", encoding="utf-8", errors="ignore") as fin, \
         open(OUTPUT_FILE, "w", encoding="utf-8") as fout:

        op = None
        db = table = None
        where_vals = {}
        set_vals = {}
        in_where = False
        skip = False

        def flush():
            nonlocal op
            if not op or skip:
                return
            try:
                if op == "INSERT":
                    cols, vals = [], []
                    for i in sorted(set_vals):
                        col = get_field(table, i)
                        cols.append(f"`{col}`")
                        vals.append(set_vals[i])
                    write_sql(
                        fout,
                        f"INSERT INTO `{db}`.`{table}` ({', '.join(cols)}) VALUES ({', '.join(vals)})"
                    )

                elif op == "UPDATE":
                    sets = []
                    wheres = []
                    for i in sorted(set_vals):
                        col = get_field(table, i)
                        sets.append(f"`{col}`={set_vals[i]}")
                    for i in sorted(where_vals):
                        col = get_field(table, i)
                        wheres.append(f"`{col}`={where_vals[i]}")
                    write_sql(
                        fout,
                        f"UPDATE `{db}`.`{table}` SET {', '.join(sets)} WHERE {' AND '.join(wheres)}"
                    )

                elif op == "DELETE":
                    wheres = []
                    for i in sorted(where_vals):
                        col = get_field(table, i)
                        wheres.append(f"`{col}`={where_vals[i]}")
                    write_sql(
                        fout,
                        f"DELETE FROM `{db}`.`{table}` WHERE {' AND '.join(wheres)}"
                    )
            except Exception as e:
                print("[ERROR]", e)
            finally:
                op = None
                where_vals.clear()
                set_vals.clear()

        for line in fin:
            line = line.rstrip()

            # 新事件开始前 flush
            if RE_INSERT.match(line) or RE_UPDATE.match(line) or RE_DELETE.match(line):
                flush()

            if m := RE_INSERT.match(line):
                op = "INSERT"
                db, table = parse_table(m.group(1))
                skip = db != TARGET_DB
                in_where = False
                continue

            if m := RE_UPDATE.match(line):
                op = "UPDATE"
                db, table = parse_table(m.group(1))
                skip = db != TARGET_DB
                in_where = True
                continue

            if m := RE_DELETE.match(line):
                op = "DELETE"
                db, table = parse_table(m.group(1))
                skip = db != TARGET_DB
                in_where = True
                continue

            if RE_WHERE.match(line):
                in_where = True
                continue

            if RE_SET.match(line):
                in_where = False
                continue

            if m := RE_COL.match(line):
                if skip or not op:
                    continue
                idx = int(m.group(1))
                val = m.group(2)
                if in_where:
                    where_vals[idx] = val
                else:
                    set_vals[idx] = val
                continue

            if RE_EVENT_END.match(line):
                flush()

        flush()

    print("DONE:", OUTPUT_FILE)


if __name__ == "__main__":
    main()
```


部署在线上服务器的MySQL数据库中有很多用来临时使用的数据库，想着清理一下，但删库一时爽，然后没刹住车，把线上一个小程序的在用唯一的数据库删掉了，并且这个数据库我没有任何备份机制。当前案例中的服务器操作系统为Ubuntu，MySQL数据库版本为8.0，且部署在Docker容器中，容器名称为mysql，下文均以此条件开展。

记一次误删MySQL数据库后根据日志进行数据恢复

编程语言

MySQL误删数据库恢复指南：通过Binlog日志恢复误删数据，详细步骤包括检查Binlog状态、复制日志文件、安装解析工具、提取建表语句和数据操作语句，最终实现数据库完整恢复。适用于Ubuntu系统下Docker部署的MySQL8.0环境。

腾讯云数据库特惠，新用户首单0.5折起


2026新春采购季

tencentdb-catalog

4核4G3M云服务器 新用户低至38元/年！

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

记一次误删MySQL数据库后根据日志进行数据恢复-腾讯云开发者社区-腾讯云

记一次误删MySQL数据库后根据日志进行数据恢复

记一次误删MySQL数据库后根据日志进行数据恢复

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐