如何编写预估数据迁移耗时脚本

本文目录导读：

1. 基础估算脚本
2. 使用示例
3. 高级功能扩展
4. 注意事项

我来分享一个预估数据迁移耗时的脚本编写方法,包含多个维度的估算策略。

基础估算脚本

#!/usr/bin/env python3
"""
数据迁移耗时预估脚本
支持多种数据源和迁移方式
"""
import time
import psutil
import subprocess
from datetime import datetime, timedelta
import argparse
import json
class MigrationEstimator:
    def __init__(self, source_type='mysql', target_type='mysql'):
        self.source_type = source_type
        self.target_type = target_type
        self.sample_data = []
        self.network_speed = self.get_network_speed()
        self.io_speed = self.get_io_speed()
    def get_network_speed(self):
        """获取网络传输速度(MB/s)"""
        try:
            # 使用ping或iperf3测试网络
            result = subprocess.run(
                ['iperf3', '-c', 'target_host', '-t', '5', '-f', 'm'],
                capture_output=True, text=True, timeout=10
            )
            # 解析结果获取带宽
            speed = self.parse_iperf_output(result.stdout)
            return speed if speed > 0 else 100  # 默认100MB/s
        except:
            # 回退方法：测试到目标主机的延迟和带宽
            try:
                subprocess.run(['ping', '-c', '3', 'target_host'], 
                             capture_output=True, timeout=10)
                # 估算带宽（简化版）
                return 50  # 假设50MB/s
            except:
                return 10  # 最保守估计10MB/s
    def get_io_speed(self):
        """获取磁盘IO速度(MB/s)"""
        try:
            # 使用dd命令测试磁盘写入速度
            result = subprocess.run(
                ['dd', 'if=/dev/zero', 'of=/tmp/test_speed', 
                 'bs=1M', 'count=1024', 'oflag=dsync'],
                capture_output=True, text=True, timeout=30
            )
            # 解析速度
            for line in result.stderr.split('\n'):
                if 'MB/s' in line or 'MB/s' in line:
                    speed = self.extract_speed(line)
                    return speed
            return 200  # 默认200MB/s
        except:
            return 100  # 保守估计
    def estimate_by_sample(self, data_size_gb, sample_size_mb=10):
        """基于抽样的估算方法"""
        print(f"[{datetime.now()}] 开始抽样测试...")
        # 生成测试数据
        test_data = self.generate_sample_data(sample_size_mb)
        # 测试传输速度
        start_time = time.time()
        self.transfer_sample(test_data)
        actual_time = time.time() - start_time
        # 估算总时间
        sample_speed = (sample_size_mb / 1024) / actual_time  # GB/s
        estimated_total_time = data_size_gb / sample_speed
        return estimated_total_time, actual_time, sample_speed
    def estimate_by_formula(self, data_size_gb, **kwargs):
        """基于公式的估算方法"""
        # 数据大小因素
        size_factor = data_size_gb
        # 网络因素
        network_factor = 1.0
        if self.network_speed < 10:
            network_factor = 3.0
        elif self.network_speed < 50:
            network_factor = 1.5
        elif self.network_speed > 100:
            network_factor = 0.8
        # IO因素
        io_factor = 1.0
        if self.io_speed < 50:
            io_factor = 2.0
        elif self.io_speed < 100:
            io_factor = 1.5
        elif self.io_speed > 500:
            io_factor = 0.7
        # 数据压缩率
        compression_factor = kwargs.get('compression', 1.0)
        # 并行度
        parallel_factor = 1.0 / kwargs.get('parallel', 1)
        # 额外因素
        overhead_factor = 1.2  # 20%开销
        estimated_time = (size_factor * network_factor * io_factor * 
                         overhead_factor * compression_factor * parallel_factor)
        return estimated_time
    def estimate_by_historical(self, historical_data):
        """基于历史数据的估算"""
        if not historical_data:
            return None
        # 计算历史平均速度
        speeds = [h['size_gb'] / h['time_hours'] for h in historical_data]
        avg_speed = sum(speeds) / len(speeds)
        return avg_speed
    def comprehensive_estimate(self, data_size_gb, **kwargs):
        """综合估算方法"""
        print("="*50)
        print("数据迁移耗时综合估算")
        print(f"数据量: {data_size_gb} GB")
        print(f"网络速度: {self.network_speed:.2f} MB/s")
        print(f"IO速度: {self.io_speed:.2f} MB/s")
        print("="*50)
        # 方法1: 公式估算
        formula_time = self.estimate_by_formula(data_size_gb, **kwargs)
        print(f"公式法估算: {self.format_time(formula_time)}")
        # 方法2: 抽样估算
        if kwargs.get('do_sample', True):
            sample_time, actual_time, speed = self.estimate_by_sample(
                data_size_gb, kwargs.get('sample_size', 10)
            )
            print(f"抽样法估算: {self.format_time(sample_time)}")
            print(f"  抽样数据量: {kwargs.get('sample_size', 10)}MB")
            print(f"  抽样耗时: {actual_time:.2f}s")
            print(f"  预估速度: {speed:.2f} GB/s")
        # 综合评分
        estimates = [formula_time]
        if kwargs.get('do_sample', True):
            estimates.append(sample_time)
        avg_time = sum(estimates) / len(estimates)
        min_time = min(estimates)
        max_time = max(estimates)
        return {
            'data_size_gb': data_size_gb,
            'estimated_time_hours': avg_time,
            'time_range': (min_time, max_time),
            'network_speed': self.network_speed,
            'io_speed': self.io_speed,
            'formula_estimate': formula_time,
            'sample_estimate': sample_time if kwargs.get('do_sample', True) else None,
            'confidence': 'high' if max_time - min_time < 0.5 * avg_time else 'medium'
        }
    def format_time(self, hours):
        """格式化时间显示"""
        if hours < 1:
            minutes = hours * 60
            return f"{minutes:.1f} 分钟"
        elif hours < 24:
            return f"{hours:.1f} 小时"
        else:
            days = hours / 24
            return f"{days:.1f} 天 ({hours:.1f} 小时)"
    def generate_sample_data(self, size_mb):
        """生成测试数据"""
        return b'x' * (size_mb * 1024 * 1024)
    def transfer_sample(self, data):
        """模拟数据传输"""
        time.sleep(0.1)  # 模拟网络延迟
        return True
    def parse_iperf_output(self, output):
        """解析iperf输出"""
        # 简化实现
        return 100
    def extract_speed(self, line):
        """从输出中提取速度"""
        # 简化实现
        return 200
def main():
    parser = argparse.ArgumentParser(description='数据迁移耗时估算工具')
    parser.add_argument('--size', type=float, required=True,
                       help='数据大小 (GB)')
    parser.add_argument('--source', type=str, default='mysql',
                       help='源数据库类型')
    parser.add_argument('--target', type=str, default='mysql',
                       help='目标数据库类型')
    parser.add_argument('--parallel', type=int, default=1,
                       help='并行度')
    parser.add_argument('--compression', type=float, default=1.0,
                       help='压缩率 (0-1)')
    parser.add_argument('--no-sample', action='store_true',
                       help='禁用抽样测试')
    parser.add_argument('--output', type=str,
                       help='输出文件路径 (JSON格式)')
    args = parser.parse_args()
    # 创建估算器
    estimator = MigrationEstimator(args.source, args.target)
    # 执行估算
    result = estimator.comprehensive_estimate(
        args.size,
        parallel=args.parallel,
        compression=args.compression,
        do_sample=not args.no_sample
    )
    # 显示结果
    print("\n" + "="*50)
    print("估算结果汇总")
    print("="*50)
    print(f"数据量: {result['data_size_gb']} GB")
    print(f"预计耗时: {estimator.format_time(result['estimated_time_hours'])}")
    print(f"时间范围: {estimator.format_time(result['time_range'][0])} - "
          f"{estimator.format_time(result['time_range'][1])}")
    print(f"置信度: {result['confidence']}")
    # 输出JSON
    if args.output:
        with open(args.output, 'w') as f:
            json.dump(result, f, indent=2, default=str)
        print(f"结果已保存到: {args.output}")
    # 生成时间线建议
    generate_timeline_suggestion(result['estimated_time_hours'])
def generate_timeline_suggestion(hours):
    """生成时间线建议"""
    print("\n" + "="*50)
    print("时间线建议")
    print("="*50)
    if hours < 1:
        print("建议: 立即执行迁移")
        print("最佳时间: 立即")
    elif hours < 4:
        print("建议: 安排在低峰时段执行")
        print("最佳时间: 凌晨2:00-6:00")
    elif hours < 12:
        print("建议: 需要周密计划")
        print("最佳时间: 周末凌晨")
        print("需要准备: 回滚方案")
    else:
        print("建议: 分批进行迁移")
        print("最佳时间: 分多个周末执行")
        print("需要准备:")
        print("  1. 详细迁移计划")
        print("  2. 数据校验方案")
        print("  3. 回滚预案")
        print("  4. 监控告警配置")
    # 显示预计完成时间
    completion_time = datetime.now() + timedelta(hours=hours)
    print(f"\n预计完成时间: {completion_time.strftime('%Y-%m-%d %H:%M:%S')}")
if __name__ == "__main__":
    main()

使用示例

基本使用

# 估算10GB数据迁移
python migration_estimator.py --size 10
# 指定并行度和压缩率
python migration_estimator.py --size 100 --parallel 4 --compression 0.5
# 保存结果到JSON
python migration_estimator.py --size 50 --output estimate_result.json

集成到Shell脚本

#!/bin/bash
# migration_estimate.sh
# 获取MySQL数据库大小
get_db_size() {
    mysql -e "
        SELECT 
            ROUND(SUM(data_length + index_length) / 1024 / 1024 / 1024, 2) AS size_gb
        FROM information_schema.tables 
        WHERE table_schema = '$1';
    " | tail -1
}
# 主逻辑
DB_NAME="your_database"
SIZE=$(get_db_size $DB_NAME)
echo "数据库 $DB_NAME 大小: ${SIZE}GB"
# 调用Python脚本估算
python3 migration_estimator.py --size $SIZE --parallel 4 --compression 0.8

高级功能扩展

分阶段估算

class PhasedMigrationEstimator:
    """分阶段迁移估算"""
    def estimate_full_migration(self, data_size_gb):
        phases = {
            'schema_migration': 0.1,  # 表结构迁移
            'full_data_migration': 0.7,  # 全量数据迁移
            'incremental_sync': 0.15,  # 增量同步
            'validation': 0.05  # 数据校验
        }
        total_time = self.get_estimated_time(data_size_gb)
        phase_times = {}
        for phase, ratio in phases.items():
            phase_times[phase] = total_time * ratio
        return phase_times

实时进度监控

class ProgressMonitor:
    def __init__(self, total_size_gb):
        self.total = total_size_gb
        self.transferred = 0
        self.start_time = time.time()
    def update(self, transferred_gb):
        self.transferred = transferred_gb
        elapsed = time.time() - self.start_time
        speed = transferred_gb / elapsed
        remaining = self.total - transferred_gb
        eta = remaining / speed
        print(f"进度: {transferred_gb:.2f}/{self.total:.2f} GB "
              f"({transferred_gb/self.total*100:.1f}%) "
              f"速度: {speed:.2f} GB/s "
              f"剩余时间: {eta:.1f} 秒")

注意事项

1. 考虑网络波动：实际速度可能变化，建议添加20%-30%的缓冲时间
2. 数据库负载：迁移过程可能影响源数据库性能
3. 数据校验时间：迁移后需要时间验证数据完整性
4. 回滚准备：为可能的失败预留回滚时间

这个脚本提供了多种估算方法，可以根据实际情况组合使用,获得更准确的时间预估。