Claude代码运行中算力中断的应对策略：从故障恢复设计到资源优化

1次阅读

共计 2432 个字符，预计需要花费 7 分钟才能阅读完成。

在分布式训练和大数据处理场景中，算力中断往往带来严重后果。例如：

训练一个 ResNet-50 模型时，第 23 小时因 Spot Instance 回收导致中断，需重新计算约 1.7TB 的中间特征
Spark 处理 10TB 日志时 Worker 节点宕机，重启后需重新 Shuffle 460GB 数据

量化损失通常包含：

时间成本：任务进度回退到最近保存点
经济成本：已消耗的计算资源成为沉没成本
机会成本：阻塞后续依赖任务

方案	优点	缺点	适用场景
Checkpoint/ 检查点	精确恢复状态	存储开销较大	长耗时计算任务
Task Retry/ 任务重试	实现简单	重复计算成本高	短时任务
Redundant Compute/ 冗余计算	零恢复延迟	资源消耗翻倍	关键实时系统

通过 100MB 模型参数的序列化测试：

import json, pickle, msgpack

# 测试数据
data = {'weights': [np.random.rand(1024,1024) for _ in range(10)],
    'metadata': {'epoch': 42, 'loss': 0.01}
}

# 序列化测试
json_size = len(json.dumps(data))     # 约 850MB
pickle_size = len(pickle.dumps(data)) # 约 410MB 
msgpack_size = len(msgpack.dumps(data)) # 约 380MB

from dataclasses import dataclass
import pickle
from pathlib import Path
from typing import Dict, Any

@dataclass
class TrainingState:
    epoch: int
    model_weights: Dict[str, Any]
    optimizer_state: Dict[str, Any]

    def save(self, path: str) -> None:
        """原子化保存状态"""
        temp_path = f"{path}.tmp"
        with open(temp_path, 'wb') as f:
            pickle.dump(self, f)
        Path(temp_path).rename(path)

    @classmethod
    def load(cls, path: str) -> 'TrainingState':
        with open(path, 'rb') as f:
            return pickle.load(f)

import functools
import signal

def checkpoint_every(interval: int):
    """每 N 次迭代自动保存检查点"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            state = kwargs.get('state')
            step = 0

            def handler(signum, frame):
                state.save(f"checkpoint_{step}.pkl")
                exit(1)

            signal.signal(signal.SIGTERM, handler)

            while True:
                result = func(*args, **kwargs)
                step += 1
                if step % interval == 0:
                    state.save(f"checkpoint_{step}.pkl")
            return result
        return wrapper
    return decorator

增量检查点：只保存变化参数（如 Adam 优化器的二阶矩估计）
异步写入：使用单独线程执行 IO 操作
压缩存储：对浮点数组采用 zstd 压缩（可节省 70% 空间）

from cryptography.fernet import Fernet

class SecureCheckpoint:
    def __init__(self, key: bytes):
        self.cipher = Fernet(key)

    def save(self, obj, path: str):
        data = pickle.dumps(obj)
        encrypted = self.cipher.encrypt(data)
        with open(path, 'wb') as f:
            f.write(encrypted)

序列化整个 Python 环境（包含 lambda 函数 / 线程状态等不可序列化对象）
忽略外部连接状态（如数据库事务未提交）
Serverless 环境使用本地临时存储（应改用 S3/EBS）

# AWS Lambda 示例
import boto3

def lambda_handler(event, context):
    s3 = boto3.client('s3')

    # 尝试加载检查点
    try:
        obj = s3.get_object(Bucket='my-bucket', Key='checkpoint.pkl')
        state = pickle.loads(obj['Body'].read())
    except s3.exceptions.NoSuchKey:
        state = InitialState()

    # 执行计算
    state = compute(state)

    # 保存新状态
    s3.put_object(
        Bucket='my-bucket',
        Key='checkpoint.pkl',
        Body=pickle.dumps(state)
    )