Python脚本如何操作数据库分区表

本文目录导读：

1. PostgreSQL分区表操作
2. MySQL分区表操作
3. 使用SQLAlchemy操作分区表
4. 自动化分区维护脚本
5. 性能监控和分析
6. 注意事项

我来详细介绍Python操作数据库分区表的方法,主要针对PostgreSQL和MySQL两种常见数据库。

PostgreSQL分区表操作

创建分区表

import psycopg2
from psycopg2.extras import RealDictCursor
def create_partitioned_table():
    """创建分区表"""
    conn = psycopg2.connect(
        host="localhost",
        port=5432,
        database="testdb",
        user="user",
        password="password"
    )
    cur = conn.cursor()
    # 创建主表（分区表）
    cur.execute("""
        CREATE TABLE IF NOT EXISTS sales (
            id SERIAL,
            sale_date DATE NOT NULL,
            amount NUMERIC(10,2),
            product_id INTEGER
        ) PARTITION BY RANGE (sale_date)
    """)
    # 创建分区
    cur.execute("""
        CREATE TABLE IF NOT EXISTS sales_2024_q1 PARTITION OF sales
        FOR VALUES FROM ('2024-01-01') TO ('2024-04-01')
    """)
    cur.execute("""
        CREATE TABLE IF NOT EXISTS sales_2024_q2 PARTITION OF sales
        FOR VALUES FROM ('2024-04-01') TO ('2024-07-01')
    """)
    conn.commit()
    cur.close()
    conn.close()
# 自动创建分区
def auto_create_partitions(start_date, end_date, interval_months=3):
    """自动创建季度分区"""
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()
    current = start_date
    while current < end_date:
        # 计算分区边界
        next_date = current + datetime.timedelta(days=interval_months * 30)
        partition_name = f"sales_{current.strftime('%Y_%m')}"
        cur.execute(f"""
            CREATE TABLE IF NOT EXISTS {partition_name} PARTITION OF sales
            FOR VALUES FROM ('{current.date()}') TO ('{next_date.date()}')
        """)
        current = next_date
    conn.commit()
    cur.close()
    conn.close()

查询分区表信息

def get_partition_info():
    """获取分区信息"""
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor(cursor_factory=RealDictCursor)
    # 查看分区表信息
    cur.execute("""
        SELECT 
            parent.relname AS parent_table,
            child.relname AS partition_name,
            pg_get_expr(child.relpartbound, child.oid) AS partition_boundary,
            pg_size_pretty(pg_relation_size(child.oid)) AS size
        FROM pg_inherits
        JOIN pg_class parent ON pg_inherits.inhparent = parent.oid
        JOIN pg_class child ON pg_inherits.inhrelid = child.oid
        WHERE parent.relname = 'sales'
        ORDER BY child.relname
    """)
    partitions = cur.fetchall()
    for partition in partitions:
        print(f"分区名: {partition['partition_name']}")
        print(f"边界: {partition['partition_boundary']}")
        print(f"大小: {partition['size']}")
        print("---")
    cur.close()
    conn.close()

MySQL分区表操作

使用mysql-connector-python

import mysql.connector
from mysql.connector import Error
def create_mysql_partitioned_table():
    """创建MySQL分区表"""
    conn = mysql.connector.connect(
        host="localhost",
        user="root",
        password="password",
        database="testdb"
    )
    cursor = conn.cursor()
    # 创建RANGE分区表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS orders (
            id INT AUTO_INCREMENT,
            order_date DATE NOT NULL,
            amount DECIMAL(10,2),
            customer_id INT,
            PRIMARY KEY (id, order_date)
        ) PARTITION BY RANGE (YEAR(order_date)) (
            PARTITION p2020 VALUES LESS THAN (2021),
            PARTITION p2021 VALUES LESS THAN (2022),
            PARTITION p2022 VALUES LESS THAN (2023),
            PARTITION p2023 VALUES LESS THAN (2024),
            PARTITION p_future VALUES LESS THAN MAXVALUE
        )
    """)
    # 创建LIST分区表
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS customers_by_region (
            id INT AUTO_INCREMENT,
            name VARCHAR(50),
            region VARCHAR(20),
            PRIMARY KEY (id, region)
        ) PARTITION BY LIST COLUMNS(region) (
            PARTITION p_north VALUES IN ('北京', '天津', '河北'),
            PARTITION p_south VALUES IN ('广东', '深圳', '海南'),
            PARTITION p_east VALUES IN ('上海', '江苏', '浙江'),
            PARTITION p_other VALUES IN ('其他')
        )
    """)
    conn.commit()
    cursor.close()
    conn.close()

管理MySQL分区

class MySQLPartitionManager:
    def __init__(self, connection_config):
        self.config = connection_config
    def add_partition(self, table_name, partition_name, value):
        """添加新分区"""
        conn = mysql.connector.connect(**self.config)
        cursor = conn.cursor()
        try:
            cursor.execute(f"""
                ALTER TABLE {table_name}
                REORGANIZE PARTITION p_future INTO (
                    PARTITION {partition_name} VALUES LESS THAN ({value}),
                    PARTITION p_future VALUES LESS THAN MAXVALUE
                )
            """)
            conn.commit()
            print(f"分区 {partition_name} 添加成功")
        except Error as e:
            print(f"添加分区失败: {e}")
            conn.rollback()
        finally:
            cursor.close()
            conn.close()
    def drop_partition(self, table_name, partition_name):
        """删除分区"""
        conn = mysql.connector.connect(**self.config)
        cursor = conn.cursor()
        try:
            cursor.execute(f"""
                ALTER TABLE {table_name} DROP PARTITION {partition_name}
            """)
            conn.commit()
            print(f"分区 {partition_name} 删除成功")
        except Error as e:
            print(f"删除分区失败: {e}")
            conn.rollback()
        finally:
            cursor.close()
            conn.close()
    def truncate_partition(self, table_name, partition_name):
        """清空分区数据"""
        conn = mysql.connector.connect(**self.config)
        cursor = conn.cursor()
        cursor.execute(f"""
            ALTER TABLE {table_name} TRUNCATE PARTITION {partition_name}
        """)
        conn.commit()
        cursor.close()
        conn.close()

使用SQLAlchemy操作分区表

ORM方式操作分区表

from sqlalchemy import create_engine, text, MetaData, Table, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Integer, String, Date, Numeric
Base = declarative_base()
class Sale(Base):
    """定义分区表模型（PostgreSQL）"""
    __tablename__ = 'sales'
    __table_args__ = {
        'postgresql_partition_by': 'RANGE (sale_date)',
        'schema': 'public'
    }
    id = Column(Integer, primary_key=True)
    sale_date = Column(Date, nullable=False)
    amount = Column(Numeric(10, 2))
    product_id = Column(Integer)
class PartitionManager:
    def __init__(self, database_url):
        self.engine = create_engine(database_url)
        self.Session = sessionmaker(bind=self.engine)
    def create_partitions(self):
        """使用原生SQL创建分区"""
        with self.engine.connect() as conn:
            # 创建分区
            conn.execute(text("""
                CREATE TABLE IF NOT EXISTS sales_2024_01 PARTITION OF sales
                FOR VALUES FROM ('2024-01-01') TO ('2024-02-01')
            """))
            conn.execute(text("""
                CREATE TABLE IF NOT EXISTS sales_2024_02 PARTITION OF sales
                FOR VALUES FROM ('2024-02-01') TO ('2024-03-01')
            """))
            conn.commit()
    def insert_data_with_partition(self, data):
        """插入数据到分区表"""
        session = self.Session()
        try:
            for record in data:
                sale = Sale(**record)
                session.add(sale)
            session.commit()
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()
    def query_specific_partition(self, partition_name):
        """查询特定分区"""
        with self.engine.connect() as conn:
            result = conn.execute(
                text(f"SELECT * FROM {partition_name} WHERE amount > :min_amount"),
                {"min_amount": 1000}
            )
            return result.fetchall()

自动化分区维护脚本

import schedule
import time
from datetime import datetime, timedelta
class AutoPartitionMaintenance:
    def __init__(self, db_type='postgresql', **connection_params):
        self.db_type = db_type
        self.connection_params = connection_params
    def create_next_month_partition(self):
        """创建下个月的分区"""
        next_month = datetime.now() + timedelta(days=30)
        partition_name = f"sales_{next_month.strftime('%Y_%m')}"
        if self.db_type == 'postgresql':
            self._create_pg_partition(partition_name, next_month)
        elif self.db_type == 'mysql':
            self._create_mysql_partition(partition_name, next_month)
    def _create_pg_partition(self, name, date):
        conn = psycopg2.connect(**self.connection_params)
        cur = conn.cursor()
        start_date = date.replace(day=1)
        if start_date.month == 12:
            end_date = start_date.replace(year=start_date.year+1, month=1)
        else:
            end_date = start_date.replace(month=start_date.month+1)
        cur.execute(f"""
            CREATE TABLE IF NOT EXISTS {name} PARTITION OF sales
            FOR VALUES FROM ('{start_date.date()}') TO ('{end_date.date()}')
        """)
        conn.commit()
        cur.close()
        conn.close()
    def cleanup_old_partitions(self, keep_months=6):
        """清理旧分区"""
        cutoff_date = datetime.now() - timedelta(days=keep_months*30)
        conn = psycopg2.connect(**self.connection_params)
        cur = conn.cursor()
        # 获取所有分区
        cur.execute("""
            SELECT child.relname, pg_get_expr(child.relpartbound, child.oid)
            FROM pg_inherits
            JOIN pg_class child ON pg_inherits.inhrelid = child.oid
            JOIN pg_class parent ON pg_inherits.inhparent = parent.oid
            WHERE parent.relname = 'sales'
        """)
        for partition_name, boundary in cur.fetchall():
            # 解析分区边界，判断是否超过保留期限
            if self._should_drop_partition(partition_name, boundary, cutoff_date):
                cur.execute(f"DROP TABLE IF EXISTS {partition_name}")
                print(f"已删除旧分区: {partition_name}")
        conn.commit()
        cur.close()
        conn.close()
    def _should_drop_partition(self, name, boundary, cutoff_date):
        """判断是否需要删除分区"""
        # 从分区名中提取日期信息
        try:
            # 假设分区名格式为 sales_2024_01
            date_str = name.split('_')[1:]
            partition_date = datetime(int(date_str[0]), int(date_str[1]), 1)
            return partition_date < cutoff_date
        except:
            return False
# 定时任务
def main():
    manager = AutoPartitionMaintenance(
        db_type='postgresql',
        host='localhost',
        database='testdb',
        user='user',
        password='password'
    )
    # 每天早上8点创建下个月分区
    schedule.every().day.at("08:00").do(manager.create_next_month_partition)
    # 每周一清理旧分区
    schedule.every().monday.at("03:00").do(
        manager.cleanup_old_partitions, keep_months=6
    )
    while True:
        schedule.run_pending()
        time.sleep(60)
if __name__ == "__main__":
    main()

性能监控和分析

def analyze_partition_performance():
    """分析分区表性能"""
    conn = psycopg2.connect(**db_config)
    cur = conn.cursor()
    # 检查分区裁剪效果
    cur.execute("""
        EXPLAIN (ANALYZE, BUFFERS)
        SELECT * FROM sales 
        WHERE sale_date >= '2024-01-01' 
        AND sale_date < '2024-02-01'
    """)
    print("查询计划:")
    for row in cur.fetchall():
        print(row[0])
    # 获取分区统计信息
    cur.execute("""
        SELECT 
            schemaname,
            tablename,
            n_live_tup as row_count,
            n_dead_tup as dead_rows,
            last_analyze,
            last_autoanalyze
        FROM pg_stat_user_tables
        WHERE tablename LIKE 'sales_%'
        ORDER BY tablename
    """)
    print("\n分区表统计:")
    for row in cur.fetchall():
        print(f"表名: {row[1]}, 行数: {row[2]}, 死元组: {row[3]}")
    cur.close()
    conn.close()

注意事项

1. 分区键选择：选择合适的分区键，通常是时间或地理字段
2. 分区数量控制：避免创建过多分区，影响数据库性能
3. 索引策略：分区表索引需要考虑全局索引和本地索引
4. 数据迁移：大型分区表可能需要特定工具进行数据迁移
5. 备份策略：分区表的备份可能需要特殊处理

示例覆盖了Python操作数据库分区表的主要场景,可以根据实际需求选择合适的方案。