중단된 작업 감지

RayCluster 매니페스트에 호스트 IP 환경 변수 추가

workerGroupSpecs RayCluster YAML의 headGroupSpec 및 모두에 다음 환경 변수를 추가합니다. 이렇게 하면 훈련 컨테이너 내에서 실행되는 Python 라이브러리가 호스트의 작업 모니터링 서비스와 통신할 수 있습니다.

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

훈련 컨테이너 이미지에 도구 키트 라이브러리 설치

PyPI 웹 사이트의 toolkit-for-ray-on-sagemaker-ai 패키지를 훈련 컨테이너 이미지에 추가합니다. 라이브러리는 SageMaker Distribution 이미지에 사전 설치되어 있습니다.

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

훈련 함수에 모니터링 추가

의미 있는 작업을 수행하기 전에 훈련 함수 SageMakerLogMonitoring.start() 내에서를 호출합니다. 이렇게 하면 훈련 시작부터 모니터링이 활성화되고 모델 로드 또는 첫 번째 순방향 패스 중에 발생하는 중단이 감지될 수 있습니다.

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),
    ),
)

cancel 작업이 중단된 작업자를 종료하면 Ray 설명서의 Ray Train의 FailureConfig가 작업자 장애를 감지하고 마지막 체크포인트에서 모든 작업자를 다시 시작합니다. 작업이 영구적으로 실패max_failures하기 전에 원하는 자동 복구 시도 횟수로 설정합니다. 이 없으면 단일 작업자 종료FailureConfig가 전체 작업에 실패합니다.

Ray Train은 다시 시작할 때 마지막으로 저장된 체크포인트에서 복원되므로 가장 최근 체크포인트 이후 작업 이상으로 훈련 진행 상황이 손실되지 않습니다. 훈련 코드가 체크포인트를 주기적으로 저장하는 경우(예: 각 에포크 경계) 작업은 마지막으로 저장된 상태에서 자동으로 재개됩니다.