脚本怎样归档分离历史数据

wen 实用脚本 2026-06-26 241

本文目录导读：

脚本怎样归档分离历史数据

按时间归档分离
数据库数据归档
文件系统层级归档
智能归档脚本（带配置和管理）
最佳实践建议

我来详细说明脚本归档分离历史数据的几种常见方法和实现方案。

按时间归档分离

1 Bash脚本示例（按日期归档）

#!/bin/bash
# 历史数据归档脚本
# 功能：按月份归档日志/数据文件
# 配置
DATA_DIR="/data/logs"
ARCHIVE_DIR="/data/archives"
RETENTION_DAYS=90   # 保留最近90天的在线数据
# 创建归档函数
archive_data() {
    local source_dir=$1
    local archive_name=$2
    # 创建归档目录
    mkdir -p "${ARCHIVE_DIR}/${archive_name}"
    # 移动过期数据到归档
    find "${source_dir}" -type f -mtime +${RETENTION_DAYS} -exec mv {} "${ARCHIVE_DIR}/${archive_name}/" \;
    # 压缩归档
    tar -czf "${ARCHIVE_DIR}/${archive_name}.tar.gz" -C "${ARCHIVE_DIR}" "${archive_name}"
    # 清理未压缩的目录
    rm -rf "${ARCHIVE_DIR}/${archive_name}"
}
# 按月归档
monthly_archive() {
    local current_date=$(date +%Y%m)
    local last_month=$(date -d "1 month ago" +%Y%m)
    echo "开始归档 ${last_month} 月份的数据..."
    archive_data "${DATA_DIR}" "archive_${last_month}"
}
# 执行归档
monthly_archive
echo "归档完成！"

2 Python脚本示例（灵活的时间窗口）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import shutil
import gzip
from datetime import datetime, timedelta
import logging
class DataArchiver:
    """数据归档分离器"""
    def __init__(self, config):
        self.source_dir = config.get('source_dir', '/data/source')
        self.archive_dir = config.get('archive_dir', '/data/archive')
        self.interval_days = config.get('interval_days', 30)
        self.file_pattern = config.get('file_pattern', '*.log')
        # 配置日志
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
    def archive_by_date(self, cutoff_date=None):
        """按日期归档历史数据"""
        if not cutoff_date:
            cutoff_date = datetime.now() - timedelta(days=self.interval_days)
        archived_count = 0
        for root, dirs, files in os.walk(self.source_dir):
            for file in files:
                file_path = os.path.join(root, file)
                file_mtime = datetime.fromtimestamp(os.path.getmtime(file_path))
                # 判断是否需要归档
                if file_mtime < cutoff_date:
                    try:
                        self._move_and_compress(file_path, file_mtime)
                        archived_count += 1
                    except Exception as e:
                        self.logger.error(f"归档失败: {file_path} - {e}")
        self.logger.info(f"完成归档: {archived_count} 个文件")
        return archived_count
    def _move_and_compress(self, file_path, file_date):
        """移动并压缩文件"""
        # 创建按年月组织的归档目录
        archive_subdir = os.path.join(
            self.archive_dir,
            str(file_date.year),
            f"{file_date.month:02d}"
        )
        os.makedirs(archive_subdir, exist_ok=True)
        # 生成归档文件名
        filename = os.path.basename(file_path)
        archive_path = os.path.join(archive_subdir, filename)
        # 移动文件
        shutil.move(file_path, archive_path)
        # 压缩文件（可选）
        self._compress_file(archive_path)
    def _compress_file(self, file_path):
        """压缩单个文件"""
        try:
            with open(file_path, 'rb') as f_in:
                with gzip.open(f"{file_path}.gz", 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
            os.remove(file_path)  # 删除原始文件
        except Exception:
            pass  # 压缩失败不影响主流程
    def list_archives(self):
        """列出所有归档"""
        for root, dirs, files in os.walk(self.archive_dir):
            for file in files:
                print(os.path.join(root, file))
# 使用示例
if __name__ == "__main__":
    config = {
        'source_dir': '/var/log/app',
        'archive_dir': '/data/archives/app_logs',
        'interval_days': 30,  # 归档30天前的数据
        'file_pattern': '*.log'
    }
    archiver = DataArchiver(config)
    # 执行归档
    archiver.archive_by_date()
    # 查看归档列表
    archiver.list_archives()

数据库数据归档

1 MySQL历史数据归档脚本

-- 创建归档存储过程
DELIMITER //
CREATE PROCEDURE archive_historical_data(
    IN archive_date DATE,
    IN archive_table VARCHAR(100)
)
BEGIN
    DECLARE archive_sql VARCHAR(500);
    DECLARE delete_sql VARCHAR(500);
    DECLARE table_name VARCHAR(200);
    -- 创建归档表
    SET table_name = CONCAT(archive_table, '_archive_', DATE_FORMAT(archive_date, '%Y%m'));
    -- 创建新表存储归档数据
    SET archive_sql = CONCAT(
        'CREATE TABLE IF NOT EXISTS ', table_name,
        ' LIKE ', archive_table
    );
    PREPARE stmt FROM archive_sql;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;
    -- 插入历史数据
    SET archive_sql = CONCAT(
        'INSERT INTO ', table_name,
        ' SELECT * FROM ', archive_table,
        ' WHERE DATE(created_at) < \'', archive_date, '\''
    );
    PREPARE stmt FROM archive_sql;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;
    -- 删除原始表中的历史数据
    SET delete_sql = CONCAT(
        'DELETE FROM ', archive_table,
        ' WHERE DATE(created_at) < \'', archive_date, '\''
    );
    PREPARE stmt FROM delete_sql;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;
    SELECT CONCAT('归档完成，共迁移 ', ROW_COUNT(), ' 条记录') AS result;
END//
DELIMITER ;
-- 调用存储过程
CALL archive_historical_data('2024-01-01', 'orders');

2 MongoDB时间序列归档

#!/usr/bin/env python3
from pymongo import MongoClient
from datetime import datetime, timedelta
import pickle
import gzip
class MongoDBArchiver:
    """MongoDB历史数据归档"""
    def __init__(self, connection_string, database):
        self.client = MongoClient(connection_string)
        self.db = self.client[database]
    def archive_collection(self, collection_name, archive_before, archive_format='file'):
        """归档历史集合"""
        collection = self.db[collection_name]
        # 查询历史数据
        query = {
            'timestamp': {'$lt': archive_before}
        }
        # 批量处理
        batch_size = 1000
        cursor = collection.find(query).batch_size(batch_size)
        archived_docs = []
        total_archived = 0
        for doc in cursor:
            archived_docs.append(doc)
            if len(archived_docs) >= batch_size:
                total_archived += self._save_batch(archived_docs, collection_name)
                archived_docs = []
        # 处理剩余的
        if archived_docs:
            total_archived += self._save_batch(archived_docs, collection_name)
        # 删除源数据
        if total_archived > 0:
            collection.delete_many(query)
        return total_archived
    def _save_batch(self, docs, collection_name):
        """保存批次数据到归档文件"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        archive_file = f"archive_{collection_name}_{timestamp}.pkl.gz"
        with gzip.open(archive_file, 'wb') as f:
            pickle.dump(docs, f)
        return len(docs)
# 使用示例
archiver = MongoDBArchiver(
    'mongodb://localhost:27017/',
    'production_db'
)
# 归档30天前的数据
archive_date = datetime.now() - timedelta(days=30)
count = archiver.archive_collection('sensor_data', archive_date)
print(f"归档了 {count} 条记录")

文件系统层级归档

1 按目录结构归档

#!/bin/bash
# 分层次归档脚本
# 将 /data/projects 下90天未修改的文件归档
ARCHIVE_ROOT="/mnt/archive"
DATA_ROOT="/data/projects"
DAYS_OLD=90
# 归档函数
archive_directory() {
    local project_name=$1
    local project_path="${DATA_ROOT}/${project_name}"
    if [ ! -d "$project_path" ]; then
        return 1
    fi
    # 按项目创建归档
    archive_file="${ARCHIVE_ROOT}/${project_name}_$(date +%Y%m%d).tar.gz"
    # 查找并归档过期文件
    find "$project_path" -type f -mtime +${DAYS_OLD} | \
    tar -czf "$archive_file" -T - 2>/dev/null
    # 删除已归档的文件
    find "$project_path" -type f -mtime +${DAYS_OLD} -delete
    echo "项目 ${project_name} 归档完成"
}
# 获取所有项目目录
for project in $(ls -d ${DATA_ROOT}/*/ 2>/dev/null); do
    project_name=$(basename "$project")
    archive_directory "$project_name"
done

智能归档脚本（带配置和管理）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import os
import shutil
import hashlib
from datetime import datetime, timedelta
class SmartArchiver:
    """智能归档管理器"""
    def __init__(self, config_file='archiver_config.json'):
        with open(config_file, 'r') as f:
            self.config = json.load(f)
        self.manifest_file = os.path.join(
            self.config.get('archive_root', '/data/archive'),
            'manifest.json'
        )
        self.manifest = self._load_manifest()
    def _load_manifest(self):
        """加载归档清单"""
        if os.path.exists(self.manifest_file):
            with open(self.manifest_file, 'r') as f:
                return json.load(f)
        return {'archives': [], 'last_archive': None}
    def _save_manifest(self):
        """保存归档清单"""
        with open(self.manifest_file, 'w') as f:
            json.dump(self.manifest, f, indent=2, default=str)
    def smart_archive(self, source_path, rules=None):
        """智能归档"""
        if not rules:
            rules = self.config.get('default_rules', {})
        for item in os.listdir(source_path):
            item_path = os.path.join(source_path, item)
            if os.path.isfile(item_path):
                self._archive_file(item_path, rules)
            elif os.path.isdir(item_path):
                self._archive_directory(item_path, rules)
        self._save_manifest()
    def _archive_file(self, file_path, rules):
        """归档单个文件"""
        file_stat = os.stat(file_path)
        file_age = (datetime.now() - datetime.fromtimestamp(file_stat.st_mtime)).days
        min_age = rules.get('min_age_days', 30)
        if file_age < min_age:
            return  # 文件不够老，跳过
        # 生成归档路径
        relative_path = os.path.relpath(file_path, self.config.get('root_path', '/'))
        archive_path = os.path.join(
            self.config.get('archive_root', '/data/archive'),
            relative_path
        )
        # 创建归档目录
        os.makedirs(os.path.dirname(archive_path), exist_ok=True)
        # 移动文件
        shutil.move(file_path, archive_path)
        # 记录归档信息
        archive_info = {
            'original_path': file_path,
            'archive_path': archive_path,
            'timestamp': datetime.now().isoformat(),
            'file_hash': self._get_file_hash(archive_path),
            'file_size': os.path.getsize(archive_path)
        }
        self.manifest['archives'].append(archive_info)
    def _get_file_hash(self, file_path):
        """计算文件哈希"""
        hasher = hashlib.sha256()
        with open(file_path, 'rb') as f:
            for chunk in iter(lambda: f.read(4096), b''):
                hasher.update(chunk)
        return hasher.hexdigest()
    def restore_archive(self, archive_id=None, target_path=None):
        """恢复归档"""
        if archive_id:
            archives = [a for a in self.manifest['archives'] 
                       if a.get('id') == archive_id]
        else:
            archives = self.manifest['archives']
        restored_count = 0
        for archive in archives:
            if target_path:
                restore_path = os.path.join(target_path, 
                    os.path.basename(archive['original_path']))
            else:
                restore_path = archive['original_path']
            shutil.copy2(archive['archive_path'], restore_path)
            restored_count += 1
        return restored_count
# 使用示例
if __name__ == "__main__":
    archiver = SmartArchiver('archiver_config.json')
    # 执行智能归档
    archiver.smart_archive('/var/log/applications')
    # 查看归档统计
    manifest = archiver.manifest
    print(f"总归档文件数: {len(manifest['archives'])}")
    print(f"最后归档时间: {manifest['last_archive']}")

最佳实践建议

1 归档策略选择

# 推荐的归档策略配置
archive_strategies = {
    'hot_data': {
        'retention_days': 30,
        'storage': 'SSD',
        'compression': False
    },
    'warm_data': {
        'retention_days': 90,
        'storage': 'HDD',
        'compression': True,
        'format': 'gzip'
    },
    'cold_data': {
        'retention_days': 365,
        'storage': 'Archive',
        'compression': True,
        'format': 'zstd',
        'encryption': True
    },
    'frozen_data': {
        'retention': 'permanent',
        'storage': 'Tape/Cloud',
        'compression': True,
        'checksum': True
    }
}

2 自动化部署（crontab）

# 每天凌晨2点执行归档
0 2 * * * /usr/local/bin/data_archiver.py --config /etc/archiver.conf
# 每周日进行完整性检查
0 3 * * 0 /usr/local/bin/archiver_audit.py --verify
# 每月1号清理过期归档
0 4 1 * * /usr/local/bin/cleanup_old_archives.sh --older-than 365

这些脚本可以根据你的具体需求进行调整,关键是要考虑：