通過.ibd文件恢復MySQL數(shù)據(jù)的全流程
更新時間：2025年11月23日 11:36:58 作者：九夜
.ibd 文件是 InnoDB 存儲引擎的核心數(shù)據(jù)文件,存儲表的行數(shù)據(jù)、索引及元信息,本文主要為大家介紹了通過.ibd文件恢復MySQL數(shù)據(jù)的全流程,需要的朋友可以參考下
.ibd 文件是 InnoDB 存儲引擎的核心數(shù)據(jù)文件，存儲表的行數(shù)據(jù)、索引及元信息。在僅有 .ibd 文件的情況下，可通過以下流程恢復數(shù)據(jù)。
一、.ibd 文件基礎(chǔ)認知
- 表空間模式
- 獨立表空間（MySQL 5.6+ 默認）：每張表對應(yīng)獨立
.ibd文件（如user.ibd），包含表的完整數(shù)據(jù)、索引及元數(shù)據(jù)，默認存儲于datadir/數(shù)據(jù)庫名/目錄。 - 系統(tǒng)表空間（舊模式）：所有表數(shù)據(jù)集中存儲于
ibdata1等共享文件，無獨立.ibd文件。
- 獨立表空間（MySQL 5.6+ 默認）：每張表對應(yīng)獨立
- 核心特性
- 二進制格式，需通過 MySQL 命令或?qū)Ｓ霉ぞ呓馕觯?/li>
- 依賴 redo 日志保障事務(wù)安全，通過 undo 日志支持 MVCC；
- 最小 I/O 單位為 16KB 頁，由區(qū)（64 頁）和段（索引對應(yīng)的區(qū)集合）組成。
二、恢復工具與流程
使用開源工具 ibd2sql 解析 .ibd 文件，支持提取表結(jié)構(gòu)（DDL）和數(shù)據(jù)（SQL）。
1. 工具準備
# 下載工具 wget https://github.com/ddcw/ibd2sql/archive/refs/heads/ibd2sql-v2.x.zip unzip ibd2sql-v2.x.zip cd ibd2sql-ibd2sql-v2.x/ # 確認 Python3 環(huán)境 python --version # 需 Python 3.x
2. 單文件解析
python3 main.py your_file.ibd --sql --ddl
執(zhí)行后生成包含表結(jié)構(gòu)和數(shù)據(jù)的 SQL 文件。
三、批量處理腳本
當存在大量 .ibd 文件時，使用以下腳本批量解析：
import os
import sys
import subprocess
from datetime import datetime
from pathlib import Path
# ===================== 配置參數(shù)（按需修改）=====================
ROOT_DIR = Path(__file__).resolve().parent
SOURCE_IBD_DIR = Path(os.path.expanduser("./input_ibd")).resolve() # 源 ibd 目錄
OUTPUT_SQL_DIR = Path(os.path.expanduser("./output_sql")).resolve() # 輸出 SQL 目錄
PYTHON_CMD = sys.executable or "python3" # 使用當前 python
MAIN_SCRIPT = ROOT_DIR / "main.py"
# ===============================================================
OUTPUT_SQL_DIR.mkdir(parents=True, exist_ok=True)
# 日志文件（macOS 下默認編碼為 UTF-8，無需額外設(shè)置）
LOG_FILE = OUTPUT_SQL_DIR / f"parse_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
def log(info, level="INFO"):
"""日志輸出（控制臺 + 文件）"""
msg = f"[{level}] [{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {info}"
print(msg)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(msg + "\n")
def parse_single_ibd(ibd_file_path: Path) -> bool:
"""調(diào)用官方 main.py 解析單個 ibd，并將 stdout 寫入同名 .sql 文件"""
table_name = ibd_file_path.stem
sql_file = OUTPUT_SQL_DIR / f"{table_name}.sql"
log(f"開始解析表：{table_name}（文件：{ibd_file_path.name}）")
if not MAIN_SCRIPT.exists():
log(f"? 未找到 main.py，期望路徑：{MAIN_SCRIPT}", level="ERROR")
return False
cmd = [
PYTHON_CMD,
str(MAIN_SCRIPT),
str(ibd_file_path),
"--sql",
"--ddl",
]
try:
with open(sql_file, "w", encoding="utf-8") as sql_fp:
proc = subprocess.run(
cmd,
stdout=sql_fp,
stderr=subprocess.PIPE,
text=True,
cwd=ROOT_DIR,
check=False,
)
if proc.returncode != 0:
log(f"? 解析失?。ㄍ顺龃a {proc.returncode}）：{proc.stderr.strip()}", level="ERROR")
if sql_file.exists():
sql_file.unlink()
return False
if proc.stderr.strip():
log(f"?? 命令警告：{proc.stderr.strip()}", level="WARNING")
log(f"? SQL 生成成功：{sql_file}")
return True
except FileNotFoundError as exc:
log(f"? 命令執(zhí)行失敗：{exc}", level="ERROR")
except Exception as exc:
log(f"? 未知異常：{exc}", level="ERROR")
return False
if __name__ == "__main__":
log("=" * 60)
log("ibd2sql 批量解析任務(wù)（macOS 版）啟動")
log(f"源 ibd 目錄：{SOURCE_IBD_DIR}")
log(f"輸出 SQL 目錄：{OUTPUT_SQL_DIR}")
log("=" * 60 + "\n")
# 遍歷所有 .ibd 文件（自動過濾非 ibd 文件）
if not SOURCE_IBD_DIR.is_dir():
log("?? 源目錄不存在，請檢查配置！", level="ERROR")
sys.exit(1)
ibd_files = sorted(SOURCE_IBD_DIR.glob("*.ibd"))
total_count = len(ibd_files)
success_count = 0
fail_count = 0
fail_tables = []
if total_count == 0:
log("?? 未發(fā)現(xiàn) ibd 文件，請檢查源目錄是否正確！", level="WARNING")
sys.exit(1)
log(f"發(fā)現(xiàn) {total_count} 個 ibd 文件，開始批量解析...\n")
for ibd_path in ibd_files:
if parse_single_ibd(ibd_path):
success_count += 1
else:
fail_count += 1
fail_tables.append(ibd_path.stem)
log("-" * 40 + "\n")
# 輸出統(tǒng)計結(jié)果
log("=" * 60)
log("批量解析任務(wù)結(jié)束")
log(f"總文件數(shù)：{total_count}")
log(f"成功解析：{success_count} 個")
log(f"解析失敗：{fail_count} 個")
if fail_tables:
log(f"失敗表名：{','.join(fail_tables)}", level="ERROR")
log(f"日志文件：{LOG_FILE}")
log("=" * 60)
四、SQL 文件修正
解析后的 SQL 可能存在格式問題，需進一步處理：
1. 移除 JSON 字段的字符集聲明
MySQL JSON 類型為二進制存儲，無需指定字符集/排序規(guī)則，腳本如下：
import argparse
import sys
from dataclasses import dataclass
from pathlib import Path
import re
# 默認輸出目錄，與 batch_ibd2sql.py 保持一致
DEFAULT_OUTPUT_SQL_DIR = Path(__file__).resolve().parent / "output_sql"
JSON_KEYWORD = re.compile(r"\bjson\b", re.IGNORECASE)
CHARSET_PATTERN = re.compile(r"\s+CHARACTER\s+SET\s+\w+", re.IGNORECASE)
COLLATE_PATTERN = re.compile(r"\s+COLLATE\s+\w+", re.IGNORECASE)
COLUMN_NAME_PATTERN = re.compile(r"`([^`]+)`")
@dataclass
class ProcessResult:
file_path: Path
columns: list[str]
modified: bool
def clean_line(line: str) -> tuple[str, str | None]:
"""
若該行定義 JSON 字段且指定了字符集/排序規(guī)則，則移除相關(guān)聲明。
返回 (新行內(nèi)容, 字段名或 None)。
"""
if not JSON_KEYWORD.search(line):
return line, None
has_charset = "CHARACTER SET" in line.upper()
has_collate = "COLLATE" in line.upper()
if not (has_charset or has_collate):
return line, None
new_line = CHARSET_PATTERN.sub("", line)
new_line = COLLATE_PATTERN.sub("", new_line)
if new_line == line:
return line, None
match = COLUMN_NAME_PATTERN.search(line)
column_name = match.group(1) if match else "<unknown>"
return new_line, column_name
def process_sql_file(file_path: Path, dry_run: bool = False) -> ProcessResult:
original = file_path.read_text(encoding="utf-8")
lines = original.splitlines()
trailing_newline = original.endswith("\n")
processed_lines = []
touched_columns: list[str] = []
for line in lines:
new_line, column = clean_line(line)
processed_lines.append(new_line)
if column is not None:
touched_columns.append(column)
modified = bool(touched_columns)
if modified and not dry_run:
new_content = "\n".join(processed_lines)
if trailing_newline:
new_content += "\n"
file_path.write_text(new_content, encoding="utf-8")
return ProcessResult(file_path=file_path, columns=touched_columns, modified=modified)
def iter_sql_files(root_dir: Path):
for path in sorted(root_dir.rglob("*.sql")):
if path.is_file():
yield path
def parse_args():
parser = argparse.ArgumentParser(
description="移除 SQL 文件中 JSON 字段的字符集/排序規(guī)則聲明（MySQL JSON 內(nèi)置二進制類型無需設(shè)定字符集）。"
)
parser.add_argument(
"-t",
"--target",
default=str(DEFAULT_OUTPUT_SQL_DIR),
help="待掃描的 SQL 目錄，默認為項目 output_sql",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="僅報告將要修改的內(nèi)容，不實際寫回文件",
)
return parser.parse_args()
def main():
args = parse_args()
target_dir = Path(args.target).expanduser().resolve()
if not target_dir.is_dir():
print(f"[ERROR] 目錄不存在：{target_dir}", file=sys.stderr)
sys.exit(1)
has_changes = False
total_files = 0
total_columns = 0
for sql_file in iter_sql_files(target_dir):
total_files += 1
result = process_sql_file(sql_file, dry_run=args.dry_run)
if result.modified:
has_changes = True
total_columns += len(result.columns)
state = "DRY-RUN" if args.dry_run else "UPDATED"
columns_str = ", ".join(result.columns)
print(f"[{state}] {sql_file} -> {columns_str}")
if not has_changes:
print(f"[INFO] 未檢測到需要處理的 JSON 字段，掃描文件數(shù)：{total_files}")
else:
mode = "dry-run" if args.dry_run else "write"
print(
f"[SUMMARY] 模式={mode}, 處理文件數(shù)={total_files}, 修正字段數(shù)={total_columns}"
)
if __name__ == "__main__":
main()
2. 修復索引注釋格式（可選）
部分表可能存在索引注釋缺少 COMMENT 關(guān)鍵字或引號的問題，可使用以下腳本修復（未完全測試，少量異常表建議手動修改）：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
修復 SQL 文件中 KEY/UNIQUE KEY/PRIMARY KEY 注釋缺少 COMMENT 關(guān)鍵字的問題
將類似: KEY `xxx` (`column`) 注釋內(nèi)容
修改為: KEY `xxx` (`column`) COMMENT '注釋內(nèi)容'
同時處理: UNIQUE KEY, KEY, PRIMARY KEY 等所有索引定義
"""
import os
import re
import sys
from pathlib import Path
from datetime import datetime
from typing import List, Tuple, Dict
# ===================== 配置參數(shù) =====================
OUTPUT_SQL_DIR = Path("./output_sql").resolve() # 輸出 SQL 目錄
# ===============================================================
# 日志文件
LOG_FILE = OUTPUT_SQL_DIR / f"fix_unique_key_comment_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
def log(info: str, level: str = "INFO"):
"""日志輸出（控制臺 + 文件）"""
msg = f"[{level}] [{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {info}"
print(msg)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(msg + "\n")
def find_key_without_comment(content: str) -> List[Tuple[int, str, str, str]]:
"""
查找所有缺少 COMMENT 關(guān)鍵字的 KEY 定義（包括 UNIQUE KEY, KEY, PRIMARY KEY）
返回: [(行號, 原始行, KEY類型, 匹配的注釋內(nèi)容), ...]
"""
issues = []
lines = content.split('\n')
for line_num, line in enumerate(lines, 1):
# 跳過已經(jīng)有 COMMENT 關(guān)鍵字的行
if 'COMMENT' in line.upper():
continue
# 檢查是否包含 KEY 定義
if 'KEY' not in line.upper():
continue
line_upper = line.upper()
match = None
key_type = None
# 按優(yōu)先級匹配：PRIMARY KEY > UNIQUE KEY > KEY
if 'PRIMARY KEY' in line_upper:
match = re.search(
r'(PRIMARY\s+KEY\s+[^)]+))\s+([^,\n]+?)(?=\s*[,)]|\s*$)',
line,
re.IGNORECASE
)
if match:
key_type = 'PRIMARY KEY'
elif 'UNIQUE KEY' in line_upper:
match = re.search(
r'(UNIQUE\s+KEY\s+[^)]+))\s+([^,\n]+?)(?=\s*[,)]|\s*$)',
line,
re.IGNORECASE
)
if match:
key_type = 'UNIQUE KEY'
elif re.search(r'\bKEY\s+', line, re.IGNORECASE):
# 普通 KEY，確保不是 UNIQUE KEY 或 PRIMARY KEY
match = re.search(
r'(\bKEY\s+[^)]+))\s+([^,\n]+?)(?=\s*[,)]|\s*$)',
line,
re.IGNORECASE
)
if match:
key_type = 'KEY'
if match and key_type:
comment_text = match.group(2).strip().rstrip(',').strip()
# 排除空注釋
if comment_text:
issues.append((line_num, line, key_type, comment_text))
return issues
def fix_key_comment(content: str) -> Tuple[str, List[Dict]]:
"""
修復 KEY 注釋問題（包括 UNIQUE KEY, KEY, PRIMARY KEY）
返回: (修復后的內(nèi)容, 修復記錄列表)
"""
lines = content.split('\n')
fixes = []
new_lines = []
for line_num, line in enumerate(lines, 1):
original_line = line
fixed = False
# 跳過已經(jīng)有 COMMENT 關(guān)鍵字的行
if 'COMMENT' in line.upper():
new_lines.append(line)
continue
# 檢查是否包含 KEY 定義
if 'KEY' not in line.upper():
new_lines.append(line)
continue
line_upper = line.upper()
match = None
key_type = None
pattern = None
# 按優(yōu)先級匹配：PRIMARY KEY > UNIQUE KEY > KEY
if 'PRIMARY KEY' in line_upper:
pattern = r'(PRIMARY\s+KEY\s+[^)]+))\s+([^,\n]+?)(?=\s*[,)]|\s*$)'
match = re.search(pattern, line, re.IGNORECASE)
if match:
key_type = 'PRIMARY KEY'
elif 'UNIQUE KEY' in line_upper:
pattern = r'(UNIQUE\s+KEY\s+[^)]+))\s+([^,\n]+?)(?=\s*[,)]|\s*$)'
match = re.search(pattern, line, re.IGNORECASE)
if match:
key_type = 'UNIQUE KEY'
elif re.search(r'\bKEY\s+', line, re.IGNORECASE):
# 普通 KEY，確保不是 UNIQUE KEY 或 PRIMARY KEY
pattern = r'(\bKEY\s+[^)]+))\s+([^,\n]+?)(?=\s*[,)]|\s*$)'
match = re.search(pattern, line, re.IGNORECASE)
if match:
key_type = 'KEY'
if match and key_type and pattern:
key_def = match.group(1)
comment_text = match.group(2).strip()
# 排除空注釋或只是空白字符
if comment_text and comment_text.strip():
# 移除末尾的逗號（如果有）
comment_text = comment_text.rstrip(',').strip()
# 檢查注釋是否已經(jīng)用引號包裹
if not (comment_text.startswith("'") and comment_text.endswith("'")):
# 轉(zhuǎn)義單引號
escaped_comment = comment_text.replace("'", "''")
# 用單引號包裹
quoted_comment = f"'{escaped_comment}'"
else:
quoted_comment = comment_text
# 使用正則替換
# 匹配: KEY ... ) 注釋內(nèi)容 [逗號或行尾]
# 替換為: KEY ... ) COMMENT '注釋內(nèi)容' [逗號或行尾]
new_line = re.sub(
pattern,
rf'\1 COMMENT {quoted_comment}',
line,
count=1,
flags=re.IGNORECASE
)
# 確保行尾的逗號被保留（如果原來有的話）
if line.rstrip().endswith(',') and not new_line.rstrip().endswith(','):
new_line = new_line.rstrip() + ','
fixes.append({
'line': line_num,
'key_type': key_type,
'original': original_line.strip(),
'fixed': new_line.strip(),
'comment': quoted_comment
})
new_lines.append(new_line)
fixed = True
if not fixed:
new_lines.append(line)
return '\n'.join(new_lines), fixes
def process_sql_file(sql_file: Path) -> Dict:
"""處理單個 SQL 文件"""
result = {
'file': str(sql_file),
'fixed': False,
'fixes': []
}
try:
with open(sql_file, 'r', encoding='utf-8') as f:
content = f.read()
# 查找問題
issues = find_key_without_comment(content)
if issues:
# 修復問題
new_content, fixes = fix_key_comment(content)
if fixes:
# 寫回文件
with open(sql_file, 'w', encoding='utf-8') as f:
f.write(new_content)
result['fixed'] = True
result['fixes'] = fixes
log(f"? 修復文件: {sql_file.name}")
for fix in fixes:
key_type = fix.get('key_type', 'KEY')
log(f" 行 {fix['line']} ({key_type}): {fix['original']}")
log(f" -> {fix['fixed']}")
else:
log(f"?? 發(fā)現(xiàn)問題但未修復: {sql_file.name}")
else:
log(f"? 無問題: {sql_file.name}")
except Exception as e:
log(f"? 處理文件失敗 {sql_file.name}: {str(e)}", level="ERROR")
result['error'] = str(e)
return result
def main():
"""主函數(shù)"""
log("=" * 60)
log("KEY/UNIQUE KEY/PRIMARY KEY 注釋修復任務(wù)啟動")
log(f"掃描目錄：{OUTPUT_SQL_DIR}")
log("=" * 60 + "\n")
if not OUTPUT_SQL_DIR.exists():
log(f"? 目錄不存在：{OUTPUT_SQL_DIR}", level="ERROR")
sys.exit(1)
# 查找所有 SQL 文件
sql_files = sorted(OUTPUT_SQL_DIR.glob("*.sql"))
total_count = len(sql_files)
if total_count == 0:
log("?? 未發(fā)現(xiàn) SQL 文件", level="WARNING")
sys.exit(1)
log(f"發(fā)現(xiàn) {total_count} 個 SQL 文件，開始掃描...\n")
# 統(tǒng)計信息
fixed_files = []
total_fixes = 0
all_fixes_detail = []
# 處理每個文件
for sql_file in sql_files:
result = process_sql_file(sql_file)
if result['fixed']:
fixed_files.append(result['file'])
total_fixes += len(result['fixes'])
# 為每個修復添加文件信息
for fix in result['fixes']:
fix['file'] = result['file']
all_fixes_detail.extend(result['fixes'])
log("-" * 40 + "\n")
# 輸出統(tǒng)計結(jié)果
log("=" * 60)
log("修復任務(wù)結(jié)束")
log(f"總文件數(shù)：{total_count}")
log(f"修復文件數(shù)：{len(fixed_files)}")
log(f"總修復數(shù)：{total_fixes}")
log("=" * 60)
# 輸出詳細修復信息
if all_fixes_detail:
log("\n詳細修復信息：")
log("=" * 60)
for fix in all_fixes_detail:
# 從文件路徑中提取文件名
file_name = Path(fix.get('file', '')).name if 'file' in fix else 'unknown'
key_type = fix.get('key_type', 'KEY')
log(f"\n文件: {file_name}")
log(f"行號: {fix['line']}")
log(f"類型: {key_type}")
log(f"原始: {fix['original']}")
log(f"修復: {fix['fixed']}")
log(f"注釋: {fix['comment']}")
log("=" * 60)
log(f"\n日志文件：{LOG_FILE}")
if __name__ == "__main__":
main()
總結(jié)
通過 ibd2sql 工具結(jié)合批量處理腳本，可高效恢復 .ibd 文件中的數(shù)據(jù)。解析后需注意修正 JSON 字段格式及索引注釋問題，少量異常表建議手動調(diào)整以確保數(shù)據(jù)完整性。
以上就是通過.ibd文件恢復MySQL數(shù)據(jù)的全流程的詳細內(nèi)容，更多關(guān)于.ibd文件恢復MySQL數(shù)據(jù)的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
下面小編就為大家?guī)硪黄斦刴ysqldump數(shù)據(jù)導出的問題。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-03-03
mysql中使用sql命令將時間戳解析成datetime類型存入
這篇文章主要介紹了mysql中使用sql命令將時間戳解析成datetime類型存入，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-11-11
這篇文章主要介紹了MySQL下載安裝配置詳細教程?附下載資源,本文通過圖文并茂的形式給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-09-09
這篇文章主要給大家介紹了關(guān)于MySQL慢查詢?nèi)罩镜淖饔煤烷_啟的相關(guān)資料，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-01-01
mysql 字符串轉(zhuǎn)數(shù)組的實現(xiàn)示例
有時候,我們需要將一個字符串拆分成一個數(shù)組,本文主要介紹了mysql 字符串轉(zhuǎn)數(shù)組的實現(xiàn)示例,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2024-01-01
MYSQL 數(shù)據(jù)庫命名與設(shè)計規(guī)范
對于MYSQL 數(shù)據(jù)庫的命名與設(shè)計，需要一定的規(guī)范，所以我們要了解和快速的掌握mysql有很多的幫助。2008-12-12
今天小編就為大家分享一篇關(guān)于MySQL常用SQL語句總結(jié)包含復雜SQL查詢，小編覺得內(nèi)容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2019-02-02
SQL高級查詢與預(yù)處理語句解析以及實戰(zhàn)練習題
預(yù)處理語句是數(shù)據(jù)庫優(yōu)化技術(shù)中的一項重要功能,它允許數(shù)據(jù)庫服務(wù)器預(yù)先解析SQL語句并生成執(zhí)行計劃,然后在后續(xù)執(zhí)行中只需傳遞參數(shù)即可重復使用,這篇文章主要介紹了SQL高級查詢與預(yù)處理語句解析以及實戰(zhàn)練習題的相關(guān)資料,需要的朋友可以參考下2025-10-10
這篇文章主要給大家介紹了關(guān)于MYSQL本地安裝以及出現(xiàn)問題的解決方法，文中通過示例代碼介紹的非常詳細，對大家學習或者使用mysql具有一定的參考學習價值，需要的朋友們下面來一起學習學習吧2019-03-03
解讀mysql主從配置及其原理分析(Master-Slave)
在windows下配置的，后面會在Linux下配置進行測試，需要配置mysql數(shù)據(jù)庫同步的朋友可以參考下。2011-05-05