本文目录导读：
我来给您介绍几种常见数据库的自动分区脚本编写方法。
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';
注意事项
- 备份策略：分区操作前一定要备份
- 监控报警：监控分区创建失败和磁盘空间
- 性能考虑：避免频繁ALTER操作影响性能
- 兼容性：不同数据库版本语法可能有差异
- 错误处理：完善错误处理和日志记录
选择适合您业务场景的分区方案,建议先从时间范围分区开始。
