怎样实现增量备份数据表数据脚本

本文目录导读：

1. 方法一：基于时间戳的增量备份
2. 方法二：基于ID的增量备份
3. 方法三：基于触发器的实时增量备份
4. 方法四：完整的备份脚本示例
5. 方法五：使用Shell脚本实现定时增量备份
6. 使用方法建议

基于时间戳的增量备份

-- 1. 创建备份表（与源表结构相同）
CREATE TABLE backup_table_name LIKE source_table_name;
-- 2. 添加更新时间字段（如果原表没有）
ALTER TABLE source_table_name ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP;
-- 3. 创建增量备份存储过程
DELIMITER //
CREATE PROCEDURE incremental_backup(IN last_backup_time TIMESTAMP)
BEGIN
    -- 备份新增和更新的数据
    INSERT INTO backup_table_name
    SELECT * FROM source_table_name
    WHERE updated_at > last_backup_time;
    -- 记录本次备份时间
    INSERT INTO backup_log (table_name, backup_time, record_count)
    SELECT 'source_table_name', NOW(), COUNT(*)
    FROM source_table_name
    WHERE updated_at > last_backup_time;
    -- 返回备份记录数
    SELECT ROW_COUNT() AS backup_records;
END //
DELIMITER ;
-- 4. 使用示例
CALL incremental_backup('2024-01-01 00:00:00');

基于ID的增量备份

-- 1. 创建备份记录表
CREATE TABLE backup_records (
    id INT AUTO_INCREMENT PRIMARY KEY,
    table_name VARCHAR(100),
    last_backup_id INT,
    backup_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 2. 创建增量备份存储过程
DELIMITER //
CREATE PROCEDURE incremental_backup_by_id(IN table_name VARCHAR(100))
BEGIN
    DECLARE last_id INT DEFAULT 0;
    DECLARE current_max_id INT;
    -- 获取上次备份的最大ID
    SELECT COALESCE(MAX(last_backup_id), 0) INTO last_id 
    FROM backup_records 
    WHERE table_name = table_name;
    -- 获取当前最大ID
    SELECT MAX(id) INTO current_max_id 
    FROM source_table_name;
    -- 备份增量数据
    INSERT INTO backup_table_name
    SELECT * FROM source_table_name
    WHERE id > last_id AND id <= current_max_id;
    -- 更新备份记录
    INSERT INTO backup_records (table_name, last_backup_id)
    VALUES (table_name, current_max_id);
    -- 返回备份记录数
    SELECT ROW_COUNT() AS backup_records;
END //
DELIMITER ;

基于触发器的实时增量备份

-- 1. 创建变更日志表
CREATE TABLE change_log (
    id INT AUTO_INCREMENT PRIMARY KEY,
    table_name VARCHAR(100),
    record_id INT,
    action ENUM('INSERT', 'UPDATE', 'DELETE'),
    old_data JSON,
    new_data JSON,
    change_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 2. 创建触发器
DELIMITER //
CREATE TRIGGER after_insert_source
AFTER INSERT ON source_table_name
FOR EACH ROW
BEGIN
    INSERT INTO backup_table_name SELECT * FROM source_table_name WHERE id = NEW.id;
    INSERT INTO change_log (table_name, record_id, action, new_data)
    VALUES ('source_table_name', NEW.id, 'INSERT', JSON_OBJECT('id', NEW.id, 'data', ROW_TO_JSON(NEW)));
END //
CREATE TRIGGER after_update_source
AFTER UPDATE ON source_table_name
FOR EACH ROW
BEGIN
    -- 更新备份表中的数据
    UPDATE backup_table_name 
    SET column1 = NEW.column1, column2 = NEW.column2, ...
    WHERE id = NEW.id;
    INSERT INTO change_log (table_name, record_id, action, old_data, new_data)
    VALUES ('source_table_name', NEW.id, 'UPDATE', 
            JSON_OBJECT('id', OLD.id, 'data', ROW_TO_JSON(OLD)),
            JSON_OBJECT('id', NEW.id, 'data', ROW_TO_JSON(NEW)));
END //
CREATE TRIGGER after_delete_source
AFTER DELETE ON source_table_name
FOR EACH ROW
BEGIN
    DELETE FROM backup_table_name WHERE id = OLD.id;
    INSERT INTO change_log (table_name, record_id, action, old_data)
    VALUES ('source_table_name', OLD.id, 'DELETE', 
            JSON_OBJECT('id', OLD.id, 'data', ROW_TO_JSON(OLD)));
END //
DELIMITER ;

完整的备份脚本示例

# Python增量备份脚本
import mysql.connector
from datetime import datetime, timedelta
import json
class IncrementalBackup:
    def __init__(self, host, user, password, database):
        self.conn = mysql.connector.connect(
            host=host,
            user=user,
            password=password,
            database=database
        )
        self.cursor = self.conn.cursor(dictionary=True)
    def backup_by_timestamp(self, source_table, backup_table, last_backup_time=None):
        """基于时间戳的增量备份"""
        if not last_backup_time:
            # 获取上次备份时间
            self.cursor.execute(f"""
                SELECT MAX(backup_time) as last_time 
                FROM backup_records 
                WHERE table_name = %s
            """, (source_table,))
            result = self.cursor.fetchone()
            last_backup_time = result['last_time'] if result else '2020-01-01'
        # 执行增量备份
        query = f"""
            INSERT INTO {backup_table}
            SELECT * FROM {source_table}
            WHERE updated_at > %s
        """
        self.cursor.execute(query, (last_backup_time,))
        affected_rows = self.cursor.rowcount
        # 记录备份日志
        self.cursor.execute("""
            INSERT INTO backup_records (table_name, backup_time, record_count)
            VALUES (%s, NOW(), %s)
        """, (source_table, affected_rows))
        self.conn.commit()
        return affected_rows
    def backup_by_id(self, source_table, backup_table):
        """基于ID的增量备份"""
        # 获取上次备份的最大ID
        self.cursor.execute(f"""
            SELECT COALESCE(MAX(last_backup_id), 0) as last_id 
            FROM backup_records 
            WHERE table_name = %s
        """, (source_table,))
        result = self.cursor.fetchone()
        last_id = result['last_id']
        # 获取当前最大ID
        self.cursor.execute(f"SELECT MAX(id) as max_id FROM {source_table}")
        result = self.cursor.fetchone()
        current_max_id = result['max_id']
        if current_max_id > last_id:
            # 执行增量备份
            query = f"""
                INSERT INTO {backup_table}
                SELECT * FROM {source_table}
                WHERE id > %s AND id <= %s
            """
            self.cursor.execute(query, (last_id, current_max_id))
            affected_rows = self.cursor.rowcount
            # 更新备份记录
            self.cursor.execute("""
                INSERT INTO backup_records (table_name, last_backup_id, record_count)
                VALUES (%s, %s, %s)
            """, (source_table, current_max_id, affected_rows))
            self.conn.commit()
            return affected_rows
        return 0
    def close(self):
        self.cursor.close()
        self.conn.close()
# 使用示例
if __name__ == "__main__":
    backup = IncrementalBackup('localhost', 'root', 'password', 'your_database')
    # 执行增量备份
    records = backup.backup_by_timestamp('source_table', 'backup_table')
    print(f"备份了 {records} 条记录")
    # 或者基于ID备份
    records = backup.backup_by_id('source_table', 'backup_table')
    print(f"备份了 {records} 条记录")
    backup.close()

使用Shell脚本实现定时增量备份

#!/bin/bash
# 数据库配置
DB_HOST="localhost"
DB_USER="root"
DB_PASS="password"
DB_NAME="your_database"
BACKUP_DIR="/backup/incremental"
LOG_FILE="/var/log/backup.log"
# 获取上次备份时间
LAST_BACKUP_FILE="${BACKUP_DIR}/last_backup_time.txt"
if [ -f "$LAST_BACKUP_FILE" ]; then
    LAST_BACKUP_TIME=$(cat $LAST_BACKUP_FILE)
else
    LAST_BACKUP_TIME="2020-01-01 00:00:00"
fi
# 执行增量备份
CURRENT_TIME=$(date '+%Y-%m-%d %H:%M:%S')
# 备份SQL语句
BACKUP_SQL="
INSERT INTO backup_table
SELECT * FROM source_table
WHERE updated_at > '$LAST_BACKUP_TIME';
"
# 执行备份
mysql -h$DB_HOST -u$DB_USER -p$DB_PASS $DB_NAME -e "$BACKUP_SQL"
BACKUP_STATUS=$?
# 记录日志
if [ $BACKUP_STATUS -eq 0 ]; then
    echo "$CURRENT_TIME - Incremental backup completed successfully" >> $LOG_FILE
    echo "$CURRENT_TIME" > $LAST_BACKUP_FILE
else
    echo "$CURRENT_TIME - Incremental backup failed" >> $LOG_FILE
fi
# 清理过期备份（保留30天）
find $BACKUP_DIR -name "*.sql" -mtime +30 -delete
echo "Backup completed: $(date)"

使用方法建议

1. 选择合适的方法：

   • 有时间戳字段：使用方法一
   • 自增ID且不会修改：使用方法二
   • 需要实时备份：使用方法三

2. 定期执行：

   # crontab配置示例（每天凌晨2点执行）
   0 2 * * * /path/to/backup_script.sh

3. 监控和告警：

   • 记录每次备份的记录数
   • 设置备份失败的告警通知
   • 定期验证备份数据的完整性

4. 优化建议：

   • 为时间戳和ID字段创建索引
   • 使用事务保证数据一致性
   • 考虑分区策略管理大数据量备份

根据你的具体需求选择合适的增量备份方案。