Detecção de trabalhos suspensos

Adicione a variável de ambiente IP do host ao seu RayCluster manifesto

Adicione a seguinte variável de ambiente em seu YAML headGroupSpec e workerGroupSpecs em seu RayCluster YAML. Isso permite que a biblioteca Python em execução dentro do contêiner de treinamento se comunique com o serviço de monitoramento de tarefas no host.

spec:
  headGroupSpec:
    template:
      spec:
        containers:
        - name: ray-head
          env:
          - name: HYPERPOD_JMA_HOST
            valueFrom:
              fieldRef:
                fieldPath: status.hostIP
  workerGroupSpecs:
  - template:
      spec:
        containers:
        - name: ray-worker
          env:
          - name: HYPERPOD_JMA_HOST
            valueFrom:
              fieldRef:
                fieldPath: status.hostIP

Instale a biblioteca do kit de ferramentas na imagem do contêiner de treinamento

Adicione o pacote https://pypi.org/project/toolkit-for-ray-on-sagemaker-ai/ toolkit-for-ray-on-sagemaker-ai do site PyPI à imagem do seu contêiner de treinamento. A biblioteca está pré-instalada nas imagens SageMaker de distribuição.

pip install toolkit-for-ray-on-sagemaker-ai

Adicione monitoramento à sua função de treinamento

SageMakerLogMonitoring.start()Entre em contato com sua função de treinamento antes de fazer qualquer trabalho significativo. Isso garante que o monitoramento esteja ativo desde o início do treinamento e possa detectar travamentos que ocorrem durante o carregamento do modelo ou na primeira passagem para frente.

from toolkit_for_ray_on_sagemaker_ai.log_monitoring import (
    SageMakerLogMonitoring,
    LogMonitorConfig,
)
from ray.train import RunConfig, FailureConfig, ScalingConfig
from ray.train.torch import TorchTrainer

def train_func():
    # Start monitoring BEFORE any meaningful work
    SageMakerLogMonitoring(config=LogMonitorConfig(
        enabled=True,
        rules=[
            {
                "name": "training_progress",
                "type": "log_pattern",
                "enabled": True,
                "log_pattern": "(Epoch|Step|Iteration) \\d+",
                "timeout_minutes": 10,
                "start_timeout_minutes": 30,
                "stop_pattern": "Training complete",
                "fault_on_match": False,
            },
            {
                "name": "oom_detection",
                "type": "log_pattern",
                "enabled": True,
                "log_pattern": "CUDA out of memory|OutOfMemoryError|OOM",
                "fault_on_match": True,
            },
        ],
        action="cancel",
    )).start()

    # ... your training loop
    for epoch in range(num_epochs):
        print(f"Epoch {epoch}")  # This output is what the rule monitors
        train_one_epoch(model, dataloader)

    print("Training complete")  # Matches stop_pattern, deactivates the rule

# Configure FailureConfig so Ray Train automatically restarts all workers
# when the cancel action terminates a hung worker.
trainer = TorchTrainer(
    train_func,
    scaling_config=ScalingConfig(num_workers=4, use_gpu=True),
    run_config=RunConfig(
        failure_config=FailureConfig(max_failures=3),
    ),
)

Quando a cancel ação encerra um trabalhador suspenso, a documentação do Ray Train FailureConfig na documentação do Ray detecta a falha do trabalhador e reinicia todos os trabalhadores a partir do último ponto de verificação. max_failuresDefina o número de tentativas de recuperação automática que você deseja antes que o trabalho falhe permanentemente. Sem issoFailureConfig, a demissão de um único trabalhador falha em todo o trabalho.

Como o Ray Train restaura a partir do último ponto de verificação salvo na reinicialização, nenhum progresso do treinamento é perdido além do trabalho desde o ponto de verificação mais recente. Se seu código de treinamento salvar pontos de verificação periodicamente (por exemplo, em cada limite de época), o trabalho será retomado automaticamente do último estado salvo.