如何编写数据表自动分区脚本

本文目录导读：

1. MySQL 自动分区脚本
2. PostgreSQL 自动分区脚本
3. Shell 脚本自动化（通用方案）
4. Python 自动化脚本
5. 最佳实践建议

我来给您介绍几种常见数据库的自动分区脚本编写方法。

MySQL 自动分区脚本

按日期范围自动分区

-- 存储过程：自动创建月分区
DELIMITER $$
CREATE PROCEDURE auto_create_partitions()
BEGIN
    DECLARE v_max_date DATE;
    DECLARE v_next_date DATE;
    DECLARE v_partition_name VARCHAR(50);
    -- 获取最大分区日期
    SELECT MAX(SUBSTRING_INDEX(PARTITION_DESCRIPTION, "'", -2))
    INTO v_max_date
    FROM INFORMATION_SCHEMA.PARTITIONS
    WHERE TABLE_NAME = 'your_table';
    -- 如果没有分区，从当前月开始
    IF v_max_date IS NULL THEN
        SET v_max_date = DATE_FORMAT(NOW(), '%Y-%m-01');
    END IF;
    -- 创建未来3个月的分区
    SET @counter = 1;
    WHILE @counter <= 3 DO
        SET v_next_date = DATE_ADD(v_max_date, INTERVAL @counter MONTH);
        SET v_partition_name = CONCAT('p', DATE_FORMAT(v_next_date, '%Y%m'));
        SET @sql = CONCAT(
            'ALTER TABLE your_table ADD PARTITION (',
            'PARTITION ', v_partition_name,
            ' VALUES LESS THAN (\'', v_next_date, '\')',
            ')'
        );
        PREPARE stmt FROM @sql;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;
        SET @counter = @counter + 1;
    END WHILE;
END$$
DELIMITER ;
-- 设置定时任务每月执行
CREATE EVENT IF NOT EXISTS monthly_partition_event
ON SCHEDULE EVERY 1 MONTH
STARTS DATE_ADD(LAST_DAY(NOW()), INTERVAL 1 DAY)
DO
CALL auto_create_partitions();

PostgreSQL 自动分区脚本

使用函数自动创建分区

-- 创建分区函数
CREATE OR REPLACE FUNCTION auto_create_partition()
RETURNS TRIGGER AS $$
DECLARE
    v_partition_name TEXT;
    v_start_date DATE;
    v_end_date DATE;
BEGIN
    -- 计算分区范围（按月分区）
    v_start_date := DATE_TRUNC('month', NEW.created_date);
    v_end_date := v_start_date + INTERVAL '1 month';
    v_partition_name := 'your_table_' || TO_CHAR(v_start_date, 'YYYYMM');
    -- 检查分区是否存在
    IF NOT EXISTS (
        SELECT 1 FROM pg_class 
        WHERE relname = v_partition_name
    ) THEN
        -- 创建新分区
        EXECUTE FORMAT(
            'CREATE TABLE %I PARTITION OF your_table
             FOR VALUES FROM (%L) TO (%L)',
            v_partition_name,
            v_start_date,
            v_end_date
        );
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
-- 创建触发器
CREATE TRIGGER auto_partition_trigger
BEFORE INSERT ON your_table
FOR EACH ROW
EXECUTE FUNCTION auto_create_partition();
-- 或使用定时任务
CREATE OR REPLACE FUNCTION scheduled_partition_creation()
RETURNS VOID AS $$
DECLARE
    v_partition_name TEXT;
    v_start_date DATE;
    v_end_date DATE;
    v_counter INTEGER := 0;
BEGIN
    -- 创建未来6个月的分区
    WHILE v_counter < 6 LOOP
        v_start_date := DATE_TRUNC('month', NOW() + (v_counter || ' month')::INTERVAL);
        v_end_date := v_start_date + INTERVAL '1 month';
        v_partition_name := 'your_table_' || TO_CHAR(v_start_date, 'YYYYMM');
        IF NOT EXISTS (
            SELECT 1 FROM pg_class 
            WHERE relname = v_partition_name
        ) THEN
            EXECUTE FORMAT(
                'CREATE TABLE %I PARTITION OF your_table
                 FOR VALUES FROM (%L) TO (%L)',
                v_partition_name,
                v_start_date,
                v_end_date
            );
        END IF;
        v_counter := v_counter + 1;
    END LOOP;
END;
$$ LANGUAGE plpgsql;
-- 创建定时任务（需要pg_cron扩展）
SELECT cron.schedule('create-partitions', '0 0 1 * *', 
                     'SELECT scheduled_partition_creation();');

Shell 脚本自动化（通用方案）

#!/bin/bash
# auto_partition.sh - 自动分区脚本
DB_HOST="localhost"
DB_USER="root"
DB_PASS="password"
DB_NAME="your_database"
TABLE_NAME="your_table"
# MySQL版本
mysql_partition() {
    local month_offset=$1
    local next_month=$(date -d "+${month_offset} month" +%Y-%m-01)
    local partition_name="p$(date -d "+${month_offset} month" +%Y%m)"
    mysql -h ${DB_HOST} -u ${DB_USER} -p${DB_PASS} ${DB_NAME} <<EOF
        ALTER TABLE ${TABLE_NAME} ADD PARTITION (
            PARTITION ${partition_name} 
            VALUES LESS THAN ('${next_month}')
        );
EOF
}
# PostgreSQL版本
psql_partition() {
    local month_offset=$1
    local start_date=$(date -d "+${month_offset} month" +%Y-%m-01)
    local end_date=$(date -d "+$((month_offset + 1)) month" +%Y-%m-01)
    local partition_name="${TABLE_NAME}_$(date -d "+${month_offset} month" +%Y%m)"
    PGPASSWORD=${DB_PASS} psql -h ${DB_HOST} -U ${DB_USER} -d ${DB_NAME} <<EOF
        CREATE TABLE IF NOT EXISTS ${partition_name} 
        PARTITION OF ${TABLE_NAME}
        FOR VALUES FROM ('${start_date}') TO ('${end_date}');
EOF
}
# 检查并创建未来3个月的分区
for i in {1..3}; do
    mysql_partition $i
    # psql_partition $i  # PostgreSQL版本
done
# 清理过期的分区（保留6个月）
cleanup_old_partitions() {
    local cutoff_date=$(date -d "-6 months" +%Y-%m-01)
    # MySQL清理
    mysql -h ${DB_HOST} -u ${DB_USER} -p${DB_PASS} ${DB_NAME} <<EOF
        SELECT CONCAT('ALTER TABLE ${TABLE_NAME} DROP PARTITION ', 
                     PARTITION_NAME, ';')
        FROM INFORMATION_SCHEMA.PARTITIONS
        WHERE TABLE_NAME = '${TABLE_NAME}'
          AND SUBSTRING_INDEX(PARTITION_DESCRIPTION, "'", -2) < '${cutoff_date}';
EOF
}
# 添加到crontab
# 0 0 1 * * /path/to/auto_partition.sh

Python 自动化脚本

#!/usr/bin/env python3
# auto_partition.py
import mysql.connector
import psycopg2
from datetime import datetime, timedelta
from dateutil.relativedelta import relativedelta
import logging
import schedule
import time
class AutoPartition:
    def __init__(self, db_type='mysql', config=None):
        self.db_type = db_type
        self.config = config or {
            'host': 'localhost',
            'user': 'root',
            'password': 'password',
            'database': 'your_database'
        }
        self.table_name = 'your_table'
        self.setup_logging()
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('partition.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def create_connection(self):
        """创建数据库连接"""
        if self.db_type == 'mysql':
            return mysql.connector.connect(**self.config)
        elif self.db_type == 'postgresql':
            return psycopg2.connect(**self.config)
    def get_month_boundaries(self, months_ahead=0):
        """获取月份边界"""
        current = datetime.now()
        target_date = current + relativedelta(months=months_ahead)
        # 月初
        start_date = target_date.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
        # 下个月初
        next_month = start_date + relativedelta(months=1)
        return start_date, next_month
    def create_mysql_partition(self, cursor, start_date, end_date):
        """创建MySQL分区"""
        partition_name = f"p{start_date.strftime('%Y%m')}"
        end_date_str = end_date.strftime('%Y-%m-%d')
        # 检查分区是否存在
        cursor.execute("""
            SELECT PARTITION_NAME 
            FROM INFORMATION_SCHEMA.PARTITIONS 
            WHERE TABLE_NAME = %s AND PARTITION_NAME = %s
        """, (self.table_name, partition_name))
        if not cursor.fetchone():
            sql = f"""
                ALTER TABLE {self.table_name} 
                ADD PARTITION (
                    PARTITION {partition_name} 
                    VALUES LESS THAN ('{end_date_str}')
                )
            """
            cursor.execute(sql)
            self.logger.info(f"Created partition: {partition_name}")
            return True
        return False
    def create_pgsql_partition(self, cursor, start_date, end_date):
        """创建PostgreSQL分区"""
        partition_name = f"{self.table_name}_{start_date.strftime('%Y%m')}"
        start_str = start_date.strftime('%Y-%m-%d')
        end_str = end_date.strftime('%Y-%m-%d')
        # 检查分区是否存在
        cursor.execute("""
            SELECT EXISTS (
                SELECT 1 FROM pg_class WHERE relname = %s
            )
        """, (partition_name,))
        if not cursor.fetchone()[0]:
            sql = f"""
                CREATE TABLE {partition_name} 
                PARTITION OF {self.table_name}
                FOR VALUES FROM ('{start_str}') TO ('{end_str}')
            """
            cursor.execute(sql)
            self.logger.info(f"Created partition: {partition_name}")
            return True
        return False
    def create_partitions(self, months_ahead=3):
        """创建未来几个月的分区"""
        conn = self.create_connection()
        cursor = conn.cursor()
        created = False
        for i in range(1, months_ahead + 1):
            start_date, end_date = self.get_month_boundaries(i)
            if self.db_type == 'mysql':
                result = self.create_mysql_partition(cursor, start_date, end_date)
            elif self.db_type == 'postgresql':
                result = self.create_pgsql_partition(cursor, start_date, end_date)
            if result:
                created = True
        if created:
            conn.commit()
            self.logger.info("Partitions created successfully")
        else:
            self.logger.info("No new partitions needed")
        cursor.close()
        conn.close()
    def cleanup_old_partitions(self, months_to_keep=6):
        """清理旧分区"""
        conn = self.create_connection()
        cursor = conn.cursor()
        cutoff_date = datetime.now() - relativedelta(months=months_to_keep)
        cutoff_str = cutoff_date.strftime('%Y-%m-%d')
        if self.db_type == 'mysql':
            cursor.execute(f"""
                SELECT PARTITION_NAME, PARTITION_DESCRIPTION
                FROM INFORMATION_SCHEMA.PARTITIONS
                WHERE TABLE_NAME = '{self.table_name}'
                  AND CAST(SUBSTRING_INDEX(PARTITION_DESCRIPTION, "'", -2) AS DATE) < '{cutoff_str}'
            """)
            for row in cursor.fetchall():
                drop_sql = f"ALTER TABLE {self.table_name} DROP PARTITION {row[0]}"
                cursor.execute(drop_sql)
                self.logger.info(f"Dropped partition: {row[0]}")
        conn.commit()
        cursor.close()
        conn.close()
    def run_scheduled(self):
        """运行定时任务"""
        self.create_partitions(months_ahead=3)
        self.cleanup_old_partitions(months_to_keep=6)
# 使用示例
if __name__ == "__main__":
    # MySQL配置
    mysql_config = {
        'host': 'localhost',
        'user': 'root',
        'password': 'password',
        'database': 'your_database'
    }
    partitioner = AutoPartition(db_type='mysql', config=mysql_config)
    # 立即执行
    partitioner.create_partitions(months_ahead=3)
    partitioner.cleanup_old_partitions(months_to_keep=6)
    # 设置定时任务（每月1号执行）
    schedule.every().month.at("00:00").do(partitioner.run_scheduled)
    # 保持运行
    while True:
        schedule.run_pending()
        time.sleep(3600)  # 每小时检查一次

最佳实践建议

分区策略选择

-- 1. 范围分区（推荐）
CREATE TABLE orders (
    order_id INT,
    order_date DATE,
    amount DECIMAL(10,2)
)
PARTITION BY RANGE (YEAR(order_date)) (
    PARTITION p2023 VALUES LESS THAN (2024),
    PARTITION p2024 VALUES LESS THAN (2025),
    PARTITION p2025 VALUES LESS THAN (2026)
);
-- 2. 哈希分区（均匀分布）
CREATE TABLE user_logs (
    user_id INT,
    action VARCHAR(50),
    created_at TIMESTAMP
)
PARTITION BY HASH (user_id) PARTITIONS 8;
-- 3. 列表分区（特定值）
CREATE TABLE area_data (
    area_id INT,
    area_name VARCHAR(50),
    data_value DECIMAL(10,2)
)
PARTITION BY LIST (area_id) (
    PARTITION p_north VALUES IN (1, 2, 3),
    PARTITION p_south VALUES IN (4, 5, 6)
);

监控和维护脚本

-- 查看分区信息
SELECT 
    TABLE_NAME,
    PARTITION_NAME,
    PARTITION_METHOD,
    PARTITION_EXPRESSION,
    TABLE_ROWS,
    DATA_LENGTH
FROM INFORMATION_SCHEMA.PARTITIONS
WHERE TABLE_NAME = 'your_table';
-- 查看分区数据量
SELECT 
    PARTITION_NAME,
    TABLE_ROWS,
    DATA_LENGTH/1024/1024 as 'Size(MB)'
FROM INFORMATION_SCHEMA.PARTITIONS
WHERE TABLE_NAME = 'your_table';

注意事项

1. 备份策略：分区操作前一定要备份
2. 监控报警：监控分区创建失败和磁盘空间
3. 性能考虑：避免频繁ALTER操作影响性能
4. 兼容性：不同数据库版本语法可能有差异
5. 错误处理：完善错误处理和日志记录

选择适合您业务场景的分区方案,建议先从时间范围分区开始。