3.3 备份恢复 — Milvus 数据安全保障

本节导读：掌握 Milvus 数据备份与恢复策略，构建完善的数据保护体系，确保在灾难发生时能够快速恢复业务连续性。

学习目标

• 理解 Milvus 数据存储结构和备份需求
• 掌握多种备份策略和实施方案
• 学会制定数据恢复计划和演练流程
• 能够处理数据丢失和故障恢复场景
• 实现自动化备份和监控机制

核心概念

1. 数据存储架构

Milvus 的数据存储采用分层设计，理解各层次的数据特点：

内存层

• 向量数据：实际向量数据缓存
• 索引数据：索引结构和中间结果
• 查询缓存：热门查询结果缓存

持久化层

• MinIO 存储：原始向量数据和元数据
• ETCD 存储：元数据和集群状态
• 日志存储：操作日志和审计记录

备份层次

• 实时备份：内存数据快照
• 定期备份：持久化数据完整备份
• 增量备份：变化数据累积备份

2. 备份策略设计

备份类型对比

RTO/RPO 目标设定

• RTO (恢复时间目标)：

  • 关键业务：< 1 小时
  • 重要业务：< 4 小时
  • 一般业务：< 24 小时

• RPO (恢复点目标)：

  • 关键业务：< 15 分钟
  • 重要业务：< 1 小时
  • 一般业务：< 24 小时

环境准备 / 前置知识

1. 备份工具栈

备份管理工具

# 备份工具清单
- Milvus 内置备份工具
- MinIO 备份工具
- ETCD 备份工具
- 自定义备份脚本
​

监控告警工具

# 监控配置
- 备份成功率监控
- 备份时间监控
- 存储空间监控
- 恢复演练监控
​

2. 备份存储规划

# 备份存储配置示例
backup_storage:
  primary_backup:
    type: "minio"
    endpoint: "backup-minio:9000"
    bucket: "milvus-backups"
    retention: "30d"
    
  secondary_backup:
    type: "s3"
    endpoint: "s3.amazonaws.com"
    bucket: "milvus-disaster-recovery"
    retention: "90d"
    
  archive_backup:
    type: "local"
    path: "/mnt/backups/archive"
    retention: "365d"
​

分步实战

步骤 1：完整备份实施

1.1 创建备份脚本

#!/usr/bin/env python3
"""
Milvus 完整备份脚本
"""
import os
import json
import time
import shutil
import subprocess
from datetime import datetime
from typing import Dict, List
import logging

class MilvusBackupManager:
    """Milvus 备份管理器"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.backup_root = config["backup_root"]
        self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.logger = self._setup_logger()
        
        # 创建备份目录
        self.backup_dir = os.path.join(self.backup_root, f"backup_{self.timestamp}")
        os.makedirs(self.backup_dir, exist_ok=True)
        
    def _setup_logger(self):
        """设置日志"""
        logger = logging.getLogger(__name__)
        logger.setLevel(logging.INFO)
        
        handler = logging.StreamHandler()
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        
        return logger
    
    def create_full_backup(self) -> Dict:
        """创建完整备份"""
        backup_info = {
            "backup_id": self.timestamp,
            "backup_type": "full",
            "start_time": datetime.now().isoformat(),
            "status": "running",
            "components": {}
        }
        
        try:
​