Python脚本如何操作数据库压缩表

本文目录导读：

1. MySQL数据库
2. PostgreSQL数据库
3. SQLite数据库
4. SQL Server数据库
5. 通用工具函数
6. 注意事项

我来介绍几种Python操作数据库压缩表的方法,主要针对主流数据库：

MySQL数据库

使用mysql-connector-python

import mysql.connector
from mysql.connector import Error
def compress_mysql_table():
    try:
        # 连接数据库
        connection = mysql.connector.connect(
            host='localhost',
            database='your_database',
            user='your_user',
            password='your_password'
        )
        if connection.is_connected():
            cursor = connection.cursor()
            # 执行OPTIMIZE TABLE压缩表
            table_name = 'your_table'
            query = f"OPTIMIZE TABLE {table_name}"
            cursor.execute(query)
            # 或者使用ANALYZE TABLE
            # cursor.execute(f"ANALYZE TABLE {table_name}")
            # 如果是InnoDB，重建表
            cursor.execute(f"ALTER TABLE {table_name} ENGINE=InnoDB")
            print(f"表 {table_name} 压缩完成")
    except Error as e:
        print(f"错误: {e}")
    finally:
        if connection.is_connected():
            cursor.close()
            connection.close()
# 使用MySQL的压缩特性 (ROW_FORMAT=COMPRESSED)
def create_compressed_table():
    connection = mysql.connector.connect(
        host='localhost',
        database='your_database',
        user='your_user',
        password='your_password'
    )
    cursor = connection.cursor()
    # 创建压缩表
    create_query = """
    CREATE TABLE IF NOT EXISTS compressed_table (
        id INT PRIMARY KEY,
        data TEXT
    ) ROW_FORMAT=COMPRESSED
    KEY_BLOCK_SIZE=8
    """
    cursor.execute(create_query)
    # 将现有表转换为压缩格式
    alter_query = """
    ALTER TABLE existing_table 
    ROW_FORMAT=COMPRESSED
    KEY_BLOCK_SIZE=8
    """
    cursor.execute(alter_query)
    connection.commit()
    cursor.close()
    connection.close()

PostgreSQL数据库

import psycopg2
from psycopg2 import sql
def compress_postgresql_table():
    try:
        # 连接数据库
        conn = psycopg2.connect(
            host="localhost",
            database="your_database",
            user="your_user",
            password="your_password"
        )
        cur = conn.cursor()
        # 使用VACUUM FULL压缩表
        table_name = "your_table"
        cur.execute(sql.SQL("VACUUM FULL {}").format(sql.Identifier(table_name)))
        # 或使用CLUSTER重新组织表
        cur.execute(sql.SQL("CLUSTER {}").format(sql.Identifier(table_name)))
        # 分析表更新统计信息
        cur.execute(sql.SQL("ANALYZE {}").format(sql.Identifier(table_name)))
        conn.commit()
        print(f"PostgreSQL表 {table_name} 压缩完成")
    except Exception as e:
        print(f"错误: {e}")
    finally:
        if conn:
            cur.close()
            conn.close()
# 查看表大小和压缩状态
def get_table_info():
    conn = psycopg2.connect(
        host="localhost",
        database="your_database",
        user="your_user",
        password="your_password"
    )
    cur = conn.cursor()
    # 查看表大小
    cur.execute("""
    SELECT 
        pg_size_pretty(pg_total_relation_size('your_table')) as total_size,
        pg_size_pretty(pg_relation_size('your_table')) as table_size,
        pg_size_pretty(pg_total_relation_size('your_table') - pg_relation_size('your_table')) as index_size
    """)
    result = cur.fetchone()
    print(f"总大小: {result[0]}")
    print(f"表大小: {result[1]}")
    print(f"索引大小: {result[2]}")
    cur.close()
    conn.close()

SQLite数据库

import sqlite3
def compress_sqlite_database():
    try:
        # 连接数据库
        conn = sqlite3.connect('your_database.db')
        cursor = conn.cursor()
        # 启用WAL模式
        cursor.execute("PRAGMA journal_mode=WAL")
        # 执行VACUUM压缩
        cursor.execute("VACUUM")
        # 重新组织表
        cursor.execute("REINDEX")
        # 更新统计信息
        cursor.execute("ANALYZE")
        conn.commit()
        print("SQLite数据库压缩完成")
    except sqlite3.Error as e:
        print(f"错误: {e}")
    finally:
        if conn:
            conn.close()
# 检查数据库状态
def check_database_status():
    conn = sqlite3.connect('your_database.db')
    cursor = conn.cursor()
    # 查看数据库大小
    cursor.execute("PRAGMA page_count")
    page_count = cursor.fetchone()[0]
    cursor.execute("PRAGMA page_size")
    page_size = cursor.fetchone()[0]
    db_size = page_count * page_size / (1024 * 1024)  # 转换为MB
    print(f"数据库大小: {db_size:.2f} MB")
    # 查看碎片率
    cursor.execute("PRAGMA freelist_count")
    freelist_count = cursor.fetchone()[0]
    print(f"空闲页数: {freelist_count}")
    conn.close()

SQL Server数据库

import pyodbc
def compress_sqlserver_table():
    try:
        # 连接数据库
        conn = pyodbc.connect(
            'DRIVER={SQL Server};'
            'SERVER=localhost;'
            'DATABASE=your_database;'
            'UID=your_user;'
            'PWD=your_password'
        )
        cursor = conn.cursor()
        # 重建索引和压缩表
        table_name = 'your_table'
        # 方法1: 使用ALTER INDEX
        cursor.execute(f"""
        ALTER INDEX ALL ON {table_name} 
        REBUILD WITH (FILLFACTOR = 80, 
                     SORT_IN_TEMPDB = ON,
                     STATISTICS_NORECOMPUTE = OFF)
        """)
        # 方法2: 使用数据压缩
        cursor.execute(f"""
        ALTER TABLE {table_name} 
        REBUILD WITH (DATA_COMPRESSION = PAGE)
        """)
        # 或压缩为ROW级别
        cursor.execute(f"""
        ALTER TABLE {table_name} 
        REBUILD WITH (DATA_COMPRESSION = ROW)
        """)
        conn.commit()
        print(f"SQL Server表 {table_name} 压缩完成")
    except Exception as e:
        print(f"错误: {e}")
    finally:
        if conn:
            cursor.close()
            conn.close()

通用工具函数

import os
import time
class DatabaseCompressor:
    def __init__(self, db_type, connection_params):
        self.db_type = db_type
        self.connection_params = connection_params
        self.connection = None
    def connect(self):
        """创建数据库连接"""
        if self.db_type == 'mysql':
            import mysql.connector
            self.connection = mysql.connector.connect(**self.connection_params)
        elif self.db_type == 'postgresql':
            import psycopg2
            self.connection = psycopg2.connect(**self.connection_params)
        elif self.db_type == 'sqlite':
            import sqlite3
            self.connection = sqlite3.connect(self.connection_params.get('database'))
        elif self.db_type == 'sqlserver':
            import pyodbc
            conn_str = f"DRIVER={{{self.connection_params['driver']}}};"
            conn_str += f"SERVER={self.connection_params['server']};"
            conn_str += f"DATABASE={self.connection_params['database']};"
            conn_str += f"UID={self.connection_params['user']};"
            conn_str += f"PWD={self.connection_params['password']}"
            self.connection = pyodbc.connect(conn_str)
    def compress_table(self, table_name):
        """压缩指定表"""
        if not self.connection:
            self.connect()
        cursor = self.connection.cursor()
        start_time = time.time()
        try:
            if self.db_type == 'mysql':
                cursor.execute(f"OPTIMIZE TABLE {table_name}")
                cursor.execute(f"ALTER TABLE {table_name} ENGINE=InnoDB")
            elif self.db_type == 'postgresql':
                cursor.execute(f"VACUUM FULL {table_name}")
                cursor.execute(f"ANALYZE {table_name}")
            elif self.db_type == 'sqlite':
                cursor.execute("VACUUM")
            elif self.db_type == 'sqlserver':
                cursor.execute(f"""
                ALTER INDEX ALL ON {table_name} 
                REBUILD WITH (DATA_COMPRESSION = PAGE)
                """)
            self.connection.commit()
            elapsed_time = time.time() - start_time
            print(f"压缩完成 - 耗时: {elapsed_time:.2f}秒")
        except Exception as e:
            print(f"压缩失败: {e}")
            self.connection.rollback()
        finally:
            cursor.close()
    def get_table_size(self, table_name):
        """获取表大小"""
        cursor = self.connection.cursor()
        if self.db_type == 'mysql':
            cursor.execute(f"""
            SELECT 
                ROUND(SUM(data_length + index_length) / 1024 / 1024, 2) as size_mb
            FROM information_schema.tables
            WHERE table_name = '{table_name}'
            """)
            return cursor.fetchone()[0]
        elif self.db_type == 'postgresql':
            cursor.execute(f"""
            SELECT pg_size_pretty(pg_total_relation_size('{table_name}'))
            """)
            return cursor.fetchone()[0]
    def close(self):
        """关闭连接"""
        if self.connection:
            self.connection.close()
# 使用示例
if __name__ == "__main__":
    # MySQL示例
    compressor = DatabaseCompressor('mysql', {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'test_db'
    })
    # 压缩前大小
    before_size = compressor.get_table_size('my_table')
    print(f"压缩前大小: {before_size} MB")
    # 压缩表
    compressor.compress_table('my_table')
    # 压缩后大小
    after_size = compressor.get_table_size('my_table')
    print(f"压缩后大小: {after_size} MB")
    compressor.close()

注意事项

1. 备份数据: 在执行压缩操作前确保有完整备份
2. 性能影响: 大型表压缩可能需要较长时间，建议在低负载时进行
3. 锁表问题: 某些压缩操作会锁表，需要考虑业务影响
4. 磁盘空间: 压缩操作需要额外的磁盘空间
5. 测试环境: 先在生产环境外的环境测试

选择合适的压缩方法取决于具体的数据库类型、表大小和业务需求。