Weights Biases Backup Recovery Strategy —

W&B Backup & Recovery

Weights Biases W&B Backup Recovery MLOps Experiment Tracking Artifacts Model Registry API Export S3 Git DVC Automation

เนื้อหาเกี่ยวข้อง — ดูเพิ่มเติมเรื่อง Qwik Resumability DevSecOps Integration

Data Type	Location	Backup Method	Frequency
Run Metrics/Config	W&B Cloud	API Export → JSON/Parquet	Daily
Model Checkpoints	W&B Artifacts	Download → S3/GCS	Per Training Run
Datasets	W&B Artifacts / DVC	DVC Push → S3/GCS	Per Version Change
Training Scripts	Git Repository	Git Push (Already Backed Up)	Per Commit
Sweep Results	W&B Cloud	API Export → JSON	Per Sweep Completion
W&B Server DB	Self-hosted MySQL/PG	pg_dump / mysqldump → S3	Daily

Backup Script

# === W&B Backup Script ===

# pip install wandb boto3 pandas

import wandb
import json
import os
from dataclasses import dataclass, asdict
from datetime import datetime

# wandb.login(key="YOUR_API_KEY")
# api = wandb.Api()

@dataclass
class BackupConfig:
    entity: str
    project: str
    backup_dir: str
    s3_bucket: str
    s3_prefix: str
    max_runs: int

config = BackupConfig(
    entity="my-team",
    project="my-project",
    backup_dir="/tmp/wandb_backup",
    s3_bucket="my-backup-bucket",
    s3_prefix="wandb-backup",
    max_runs=1000
)

def backup_runs(api, config):
    """Backup all runs from a W&B project"""
    runs = api.runs(f"{config.entity}/{config.project}",
                    per_page=50, order="-created_at")
    backup_data = []
    for run in runs:
        run_data = {
            "id": run.id,
            "name": run.name,
            "state": run.state,
            "config": dict(run.config),
            "summary": dict(run.summary._json_dict),
            "created_at": run.created_at,
            "tags": run.tags,
            "notes": run.notes,
        }
        backup_data.append(run_data)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"runs_backup_{timestamp}.json"
    filepath = os.path.join(config.backup_dir, filename)
    os.makedirs(config.backup_dir, exist_ok=True)
    with open(filepath, 'w') as f:
        json.dump(backup_data, f, indent=2, default=str)
    print(f"Backed up {len(backup_data)} runs to {filepath}")
    return filepath

def backup_artifacts(api, config, artifact_name, version="latest"):
    """Download artifact from W&B"""
    artifact = api.artifact(
        f"{config.entity}/{config.project}/{artifact_name}:{version}")
    download_dir = os.path.join(config.backup_dir, "artifacts", artifact_name)
    artifact.download(root=download_dir)
    print(f"Downloaded {artifact_name}:{version} to {download_dir}")

print("=== W&B Backup Config ===")
for k, v in asdict(config).items():
    print(f"  {k}: {v}")