本文目录导读：
MySQL 统计脚本
统计所有表的数据量
-- 统计所有表的行数（精确）
SELECT
TABLE_NAME,
TABLE_ROWS,
DATA_LENGTH/1024/1024 AS 'Data Size (MB)',
INDEX_LENGTH/1024/1024 AS 'Index Size (MB)'
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'your_database_name'
ORDER BY TABLE_ROWS DESC;
生成统计所有表的SQL
-- 生成统计SQL
SELECT CONCAT(
'SELECT "', table_name, '" AS table_name, COUNT(*) AS row_count FROM ',
table_name, ' UNION ALL '
) AS query_part
FROM information_schema.TABLES
WHERE TABLE_SCHEMA = 'your_database_name';
-- 然后复制结果去掉最后一个UNION ALL执行
Shell脚本批量统计
#!/bin/bash
# mysql_count.sh
HOST="localhost"
USER="root"
PASS="password"
DB="your_database"
mysql -h$HOST -u$USER -p$PASS -D$DB -e "
SELECT
table_name,
engine,
table_rows,
ROUND(data_length/1024/1024, 2) AS 'data_size_mb',
ROUND(index_length/1024/1024, 2) AS 'index_size_mb'
FROM information_schema.tables
WHERE table_schema = '$DB'
ORDER BY table_rows DESC;
"
PostgreSQL 统计脚本
-- 统计所有表的数据量
SELECT
schemaname,
tablename,
n_live_tup AS row_count,
pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS total_size
FROM pg_stat_user_tables
WHERE schemaname = 'public'
ORDER BY n_live_tup DESC;
-- 精确统计（推荐）
DO $$
DECLARE
tbl RECORD;
total BIGINT;
BEGIN
FOR tbl IN
SELECT tablename FROM pg_tables
WHERE schemaname = 'public'
LOOP
EXECUTE 'SELECT COUNT(*) FROM ' || tbl.tablename INTO total;
RAISE NOTICE 'Table: %, Rows: %', tbl.tablename, total;
END LOOP;
END $$;
SQL Server 统计脚本
-- 统计所有表的数据量
SELECT
t.name AS table_name,
p.rows AS row_count,
ROUND(SUM(a.total_pages) * 8 / 1024, 2) AS 'total_size_mb'
FROM sys.tables t
INNER JOIN sys.indexes i ON t.object_id = i.object_id
INNER JOIN sys.partitions p ON i.object_id = p.object_id AND i.index_id = p.index_id
INNER JOIN sys.allocation_units a ON p.partition_id = a.container_id
WHERE t.is_ms_shipped = 0
GROUP BY t.name, p.rows
ORDER BY p.rows DESC;
Oracle 统计脚本
-- 统计所有表的数据量
SELECT
table_name,
num_rows,
ROUND(blocks * 8 / 1024, 2) AS 'size_mb'
FROM user_tables
ORDER BY num_rows DESC;
-- 需要先收集统计信息
BEGIN
DBMS_STATS.GATHER_SCHEMA_STATS(
ownname => 'YOUR_SCHEMA',
estimate_percent => 100
);
END;
/
Python 通用脚本
# count_db_data.py
import mysql.connector
import psycopg2
import pandas as pd
def count_mysql_tables(config):
"""统计MySQL所有表数据量"""
conn = mysql.connector.connect(**config)
cursor = conn.cursor()
cursor.execute("""
SELECT table_name, table_rows,
ROUND(data_length/1024/1024, 2) as data_mb,
ROUND(index_length/1024/1024, 2) as index_mb
FROM information_schema.tables
WHERE table_schema = %s
ORDER BY table_rows DESC
""", (config['database'],))
tables = cursor.fetchall()
conn.close()
return tables
def count_all_tables_exact(conn_info):
"""精确统计所有表行数"""
conn = mysql.connector.connect(**conn_info)
cursor = conn.cursor()
cursor.execute("SHOW TABLES")
tables = [row[0] for row in cursor.fetchall()]
results = []
for table in tables:
cursor.execute(f"SELECT COUNT(*) FROM {table}")
count = cursor.fetchone()[0]
results.append((table, count))
print(f"Table: {table}, Rows: {count:,}")
conn.close()
return results
# 使用示例
config = {
'host': 'localhost',
'user': 'root',
'password': 'password',
'database': 'your_database'
}
# 执行统计
tables_info = count_mysql_tables(config)
for table in tables_info:
print(f"Table: {table[0]:30} Rows: {table[1]:>10,} Data: {table[2]:>8.2f}MB Index: {table[3]:>8.2f}MB")
批量统计脚本
Shell脚本 - 多数据库统计
#!/bin/bash
# batch_db_count.sh
databases=("db1" "db2" "db3")
USER="root"
PASS="password"
for db in "${databases[@]}"; do
echo "========== Database: $db =========="
mysql -u$USER -p$PASS -D$db -e "
SELECT
CONCAT(table_schema, '.', table_name) AS 'Table',
table_rows AS 'Rows',
ROUND(data_length/1024/1024, 2) AS 'Data(MB)'
FROM information_schema.tables
WHERE table_schema = '$db'
ORDER BY table_rows DESC;
"
done
实时监控脚本
# monitor_db_size.py
import time
import mysql.connector
from datetime import datetime
def monitor_table_growth(config, interval=3600):
"""实时监控表数据增长"""
conn = mysql.connector.connect(**config)
cursor = conn.cursor()
while True:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
cursor.execute("""
SELECT table_name, table_rows
FROM information_schema.tables
WHERE table_schema = %s
ORDER BY table_rows DESC
""", (config['database'],))
print(f"\n=== 数据统计 {timestamp} ===")
for table in cursor.fetchall():
print(f"{table[0]:30} {table[1]:>10,} rows")
print(f"\n等待 {interval} 秒后重新统计...")
time.sleep(interval)
conn.close()
使用建议
- 性能考虑：对大型表使用
COUNT(*)可能很慢，考虑使用information_schema或定期更新统计信息 - 精确性：
information_schema中的TABLE_ROWS是估计值，需要精确值请使用COUNT(*) - 自动化：可以设置cron job定期执行统计脚本
- 日志记录：建议将统计结果记录到日志或专门统计表中
选择哪种方法取决于您的数据库类型、数据量大小和精确度要求，对于大多数场景，information_schema方法已经足够。
