前言
「收集汇聚表信息」翻译成工程语言,就是生成一份数据字典:这个库里有哪些表、每张表有几个字段、字段类型和注释是什么、索引怎么建的、大致占多少空间。它最常见的三个用途是:写文档、做数据治理盘点、以及给别人交接时快速摸底。
先说两个必须先纠正的认识。
第一个误解:information_schema.TABLES里的TABLE_ROWS是精确行数。不是。MySQL 官方文档明确写了:对 InnoDB 这类存储引擎,这个值只是 SQL 优化用的粗略估算值;DATA_LENGTH同理,对 InnoDB 是「近似分配的簇索引空间」。所以它可以用来做容量趋势判断,不能用来做对账或报表。要精确行数只能SELECT COUNT(*)。
第二个误解:这些元数据查询「不费资源,随便跑」。在表数量很多的实例上,全库扫描information_schema可能相当慢,还可能影响其他会话。收集任务要限定库名、避开业务高峰期、必要时加上超时。
本文用 PyMySQL 从information_schema取四种信息:库表清单、列定义、索引、表级统计,最后汇聚成一份 JSON / Markdown 数据字典。所有条件值都走参数化。
一、取元数据的三种方式
| 方式 | 写法 | 适合场景 | 局限 |
|---|
information_schema视图 | SELECT ... FROM information_schema.TABLES WHERE TABLE_SCHEMA = %s | 程序化收集、批量导出、可加条件过滤 | 字段多,需要自己拼接展示 |
SHOW系列 | SHOW TABLES、SHOW FULL COLUMNS FROM t、SHOW INDEX FROM t | 人工在客户端里快速看 | 一次只针对一张表,程序里要用SHOW的结果再解析 |
SHOW CREATE TABLE | SHOW CREATE TABLE t | 要拿到可复制的建表 DDL | 返回的是整段文本,程序难以结构化利用 |
结论:要程序化「汇聚」,就用information_schema;要人眼看,就用SHOW。两者不冲突,但别用SHOW去拼一个批量脚本——你会陷入「逐表发命令 + 解析文本」的麻烦里。
一个实用技巧:TABLE_SCHEMA = DATABASE()可以自动匹配当前连接所在的库,省掉手填库名。但库名如果来自外部输入,一定要走参数化,不能拼字符串。
二、库表清单与表级统计
# 适用于 Python 3.9+
import os
import pymysql
from pymysql.cursors import DictCursor
conn = pymysql.connect(
host=os.environ.get("MYSQL_HOST", "127.0.0.1"),
port=3306,
user=os.environ["MYSQL_USER"],
password=os.environ["MYSQL_PASSWORD"],
database=os.environ.get("MYSQL_DB", "school"),
charset="utf8mb4",
cursorclass=DictCursor,
connect_timeout=5,
)
SQL_TABLES = """
SELECT TABLE_NAME, ENGINE, TABLE_ROWS, DATA_LENGTH, INDEX_LENGTH,
TABLE_COLLATION, TABLE_COMMENT, CREATE_TIME
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = %s AND TABLE_TYPE = 'BASE TABLE'
ORDER BY TABLE_NAME
"""
db_name = os.environ.get("MYSQL_DB", "school")
with conn.cursor() as cur:
cur.execute(SQL_TABLES, (db_name,))
tables = cur.fetchall()
for t in tables:
size_kb = ((t["DATA_LENGTH"] or 0) + (t["INDEX_LENGTH"] or 0)) / 1024
print(f"{t['TABLE_NAME']:<24}{t['ENGINE']:<10}"
f"约 {t['TABLE_ROWS'] or 0:>8} 行 {size_kb:>10.1f} KB {t['TABLE_COMMENT'] or ''}")
conn.close()几个字段的含义与陷阱:
TABLE_TYPE = 'BASE TABLE'用来排除视图。视图也在TABLES里,但字段含义不同,混在一起统计会不准确。如果是分区表,分区会以BASE TABLE之外的类型另列,注意你的 MySQL 版本对应的取值。TABLE_ROWS:InnoDB 下是估算值,如上所述。DATA_LENGTH+INDEX_LENGTH:数据与索引的近似占用。判断「哪张表最大」够用,做容量审计不够。TABLE_COMMENT:建表时写的注释。如果团队不写注释,这一列会大面积是空的——数据字典的价值有一半来自注释,评审时值得专门提出来。CREATE_TIME对 InnoDB 表通常只在表创建时有值,很多场景下会是NULL,不要拿它当「最后修改时间」。
注意上面代码里的db_name是单独用变量保存的,不是从连接对象上读的。有些驱动会在连接建立后把连接对象上保存的库名内部转成字节串,直接拿去拼 SQL 或做字符串比较会得到b'school'这种结果。库名自己留一份,比从连接对象上反查更可靠。
三、列定义:数据字典的主体
列级信息是数据字典里最有用的部分:字段名、类型、是否可空、默认值、键标记、注释。
# 适用于 Python 3.9+(接上一节的 conn 和 db_name)
SQL_COLUMNS = """
SELECT TABLE_NAME, COLUMN_NAME, ORDINAL_POSITION, COLUMN_TYPE,
IS_NULLABLE, COLUMN_DEFAULT, COLUMN_KEY, EXTRA, COLUMN_COMMENT
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = %s
ORDER BY TABLE_NAME, ORDINAL_POSITION
"""
with conn.cursor() as cur:
cur.execute(SQL_COLUMNS, (db_name,))
columns = cur.fetchall()
# 按表分组,生成一份可直接看的文本字典
from collections import defaultdict
by_table = defaultdict(list)
for c in columns:
by_table[c["TABLE_NAME"]].append(c)
for table, cols in by_table.items():
print(f"\n## {table}")
print(f"{'#':<4}{'列名':<20}{'类型':<18}{'可空':<6}{'键':<5}注释")
for c in cols:
print(f"{c['ORDINAL_POSITION']:<4}{c['COLUMN_NAME']:<20}"
f"{c['COLUMN_TYPE']:<18}{c['IS_NULLABLE']:<6}"
f"{(c['COLUMN_KEY'] or '-'):<5}{c['COLUMN_COMMENT'] or ''}")字段释义:
ORDINAL_POSITION是列在表里的序位,从 1 开始。排序要靠它,不要靠列名的字母序——字母序会把物理顺序打乱,看的人会误解语义。COLUMN_TYPE是「完整类型」,例如varchar(50)、int(10) unsigned、enum('M','F')。而DATA_TYPE只有基础类型名(varchar、int、enum)。展示用COLUMN_TYPE,判断归类用DATA_TYPE。COLUMN_KEY:PRI主键、UNI唯一索引的首列、MUL非唯一索引的首列、空表示没有索引。注意它只标记「在某个索引里的首列」,组合索引的后续列不会在这里标出来——完整的索引信息要看STATISTICS。EXTRA常见值有auto_increment(自增)、DEFAULT_GENERATED(有生成表达式)、on update CURRENT_TIMESTAMP(更新时自动刷新时间)。COLUMN_DEFAULT在 MySQL 8.0 中,对有表达式的默认值会以表达式文本形式显示。
四、索引信息
# 适用于 Python 3.9+
SQL_INDEXES = """
SELECT TABLE_NAME, INDEX_NAME, NON_UNIQUE, SEQ_IN_INDEX,
COLUMN_NAME, CARDINALITY, INDEX_TYPE
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = %s
ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX
"""
with conn.cursor() as cur:
cur.execute(SQL_INDEXES, (db_name,))
indexes = cur.fetchall()
current = None
for idx in indexes:
key = (idx["TABLE_NAME"], idx["INDEX_NAME"])
if key != current:
current = key
unique = "唯一" if idx["NON_UNIQUE"] == 0 else "普通"
print(f"\n[{idx['TABLE_NAME']}] {idx['INDEX_NAME']} ({unique}, {idx['INDEX_TYPE']})")
print(f" {idx['SEQ_IN_INDEX']}. {idx['COLUMN_NAME']}")理解这张表的三个关键点:
- 一行 = 索引的一个列位。组合索引
(a, b)会产生两行,SEQ_IN_INDEX分别是 1 和 2。判断一个索引是不是组合索引,就看同一INDEX_NAME下有没有多行。 NON_UNIQUE = 0表示唯一索引。这个名字是反的,容易看错。CARDINALITY是估算的基数(不重复值个数的估计)。它由统计信息得来,可能过时,可以用ANALYZE TABLE刷新。用它判断「索引有没有被用上」「选择性高不高」只能做粗略参考,不要当精确指标。
五、汇聚成一份数据字典
把上面三类结果合起来,导出成 JSON,方便后续做比对(比如上线前后 diff 出「结构变了哪些」)。
# 适用于 Python 3.9+
import json
def collect_dictionary(conn, schema):
"""把库的表 / 列 / 索引 / 统计收进一个嵌套字典。"""
result = {}
with conn.cursor() as cur:
cur.execute(SQL_TABLES, (schema,))
for t in cur.fetchall():
result[t["TABLE_NAME"]] = {
"engine": t["ENGINE"],
"collation": t["TABLE_COLLATION"],
"comment": t["TABLE_COMMENT"] or "",
"approx_rows": t["TABLE_ROWS"],
"approx_bytes": (t["DATA_LENGTH"] or 0) + (t["INDEX_LENGTH"] or 0),
"columns": [],
"indexes": {},
}
cur.execute(SQL_COLUMNS, (schema,))
for c in cur.fetchall():
table = result.get(c["TABLE_NAME"])
if table is None: # 视图等不在表清单里的对象,跳过
continue
table["columns"].append({
"name": c["COLUMN_NAME"],
"type": c["COLUMN_TYPE"],
"nullable": c["IS_NULLABLE"] == "YES",
"default": c["COLUMN_DEFAULT"],
"key": c["COLUMN_KEY"],
"extra": c["EXTRA"],
"comment": c["COLUMN_COMMENT"] or "",
})
cur.execute(SQL_INDEXES, (schema,))
for i in cur.fetchall():
table = result.get(i["TABLE_NAME"])
if table is None:
continue
entry = table["indexes"].setdefault(
i["INDEX_NAME"],
{"unique": i["NON_UNIQUE"] == 0, "type": i["INDEX_TYPE"], "columns": []},
)
entry["columns"].append(i["COLUMN_NAME"])
return result
if __name__ == "__main__":
conn = pymysql.connect(
host="127.0.0.1", user="app", password="app_password",
database="school", charset="utf8mb4",
cursorclass=DictCursor,
)
try:
dictionary = collect_dictionary(conn, "school")
with open("schema_dump.json", "w", encoding="utf-8") as fp:
json.dump(dictionary, fp, ensure_ascii=False, indent=2)
print(f"已导出 {len(dictionary)} 张表的定义")
finally:
conn.close()注意ensure_ascii=False:不写这一项,中文注释会被转义成\uXXXX,虽然数据没错,但人工看 JSON 时完全不可读。同时open()必须显式encoding="utf-8",否则在 Windows 上会按系统默认编码(GBK)写文件,跨平台后出现乱码。
用同样的思路可以再加一层:把两次采集的 JSON 做 diff,就能得到「新增了哪些列」「哪个索引不见了」的结构变更报告。这比人工比对建表语句可靠得多。
常见坑点
1. 把TABLE_ROWS当精确行数
❌print("这张表有", row["TABLE_ROWS"], "条记录")——InnoDB 下这是估算值,可能偏差很大。
✅ 展示时明确标注「约」;需要精确值时对目标表执行SELECT COUNT(*)。
2. 把库名拼进 SQL
❌cur.execute(f"SELECT ... FROM information_schema.TABLES WHERE TABLE_SCHEMA = '{schema}'")——库名来自参数时就是注入入口。
✅cur.execute("... WHERE TABLE_SCHEMA = %s", (schema,))。
3. 忘了排除视图
❌ 只按TABLE_SCHEMA过滤,视图、分区被一并算进表数量,统计结果对不上。
✅ 明确加上TABLE_TYPE = 'BASE TABLE'(需要视图时另开一条查询)。
4. 用DATA_TYPE展示类型
❌ 输出varchar、int、date,看不到长度和 unsigned,读者无法据此建表。
✅ 展示用COLUMN_TYPE(varchar(50)、int unsigned),只有做类型归类时才用DATA_TYPE。
5. 用COLUMN_KEY判断「这一列有没有索引」
❌ 看到组合索引的第二列COLUMN_KEY是空,就认为它没索引,于是又建了一个单列索引。
✅ 索引完整性信息只在information_schema.STATISTICS里,COLUMN_KEY只标首列。
6. 把CARDINALITY当精确去重数
❌ 用CARDINALITY直接算「去重用户数」——它是估算值,还可能过时。
✅ 只把它当选择性的粗略参考;精确值用COUNT(DISTINCT col)。
7. 导出 JSON 没写ensure_ascii=False和encoding
❌json.dump(data, open("d.json", "w"))——中文注释变\uXXXX,且 Windows 上默认 GBK 编码写出会炸或乱码。
✅open("d.json", "w", encoding="utf-8")配合json.dump(..., ensure_ascii=False, indent=2)。
8. 在业务高峰期全库扫元数据
❌ 定时任务每 5 分钟对上百张表的实例做一次全量information_schema扫描。
✅ 限定库名、降低频率、避开高峰;必要时拆分成按需采集。
总结
| 采集目标 | 数据来源 | 关键注意点 |
|---|
| 表清单 | information_schema.TABLES | 用TABLE_TYPE = 'BASE TABLE'排除视图 |
| 容量/行数 | TABLES的DATA_LENGTH、INDEX_LENGTH、TABLE_ROWS | InnoDB 下全是近似值,只能做趋势 |
| 列定义 | information_schema.COLUMNS | 展示用COLUMN_TYPE;按ORDINAL_POSITION排序 |
| 索引 | information_schema.STATISTICS | 一行一个列位;NON_UNIQUE = 0才是唯一索引 |
| 导出 | json.dump到文件 | encoding="utf-8"+ensure_ascii=False |
| 安全 | 全程参数化 | 库名等条件值走%s,不要拼字符串 |
information_schema是一套标准的、只读的元数据视图,用 Python 读它的价值在于可重复、可对比、可自动化。把「采集 → 导出 → diff」做成一个小工具,数据字典就不再是一次性文档,而成了能纳入流程的检查项。