Utilizzo dei piani di formazione per lavori SageMaker di formazione

Puoi utilizzare un piano di SageMaker formazione per i tuoi lavori di formazione specificando il piano di tua scelta al momento della creazione di un lavoro di formazione.

Se la capacità richiesta non è immediatamente disponibile per un job di addestramento, il processo rimane in attesa finché la capacità non si libera, finché non viene raggiunta la condizione StoppingCondition oppure finché il processo non è restato in stato Pending per due giorni, a seconda di ciò che si verifica per primo. Se la condizione di arresto è soddisfatta, il processo viene arrestato. Se un processo resta in sospeso per due giorni, viene terminato con InsufficientCapacityError.

Checkpoint del job di addestramento

Quando utilizzi i piani di SageMaker formazione per i tuoi lavori di SageMaker formazione, assicurati di implementare i checkpoint nello script di formazione. Questa operazione consente di salvare i progressi dell’addestramento prima della scadenza di una capacità riservata. I checkpoint sono particolarmente importanti quando si lavora con capacità riservate, perché consentono di riprendere l’addestramento dall’ultimo punto salvato se il processo viene interrotto tra due capacità riservate o quando il piano di addestramento raggiunge la data di fine.

Per farlo, puoi utilizzare la variabile di ambiente SAGEMAKER_CURRENT_CAPACITY_BLOCK_EXPIRATION_TIMESTAMP. Questa variabile aiuta a determinare quando avviare il processo di checkpoint. Incorporando questa logica nello script di addestramento, ti assicuri che i progressi del modello vengano salvati a intervalli regolari.

Ecco un esempio di come puoi implementare questa logica di checkpoint nel tuo script di addestramento in Python:

import os
import time
from datetime import datetime, timedelta

def is_close_to_expiration(threshold_minutes=30):
    # Retrieve the expiration timestamp from the environment variable
    expiration_time_str = os.environ.get('SAGEMAKER_CURRENT_CAPACITY_BLOCK_EXPIRATION_TIMESTAMP', '0')
    
    # If the timestamp is not set (default '0'), return False
    if expiration_time_str == '0':
        return False
    
    # Convert the timestamp string (in milliseconds) to a datetime object
    expiration_time = datetime(1970, 1, 1) + timedelta(milliseconds=int(expiration_time_str))
    
    # Calculate the time difference between now and the expiration time
    time_difference = expiration_time - datetime.now()
    
    # Return True if we're within the threshold time of expiration
    return time_difference < timedelta(minutes=threshold_minutes)

def start_checkpointing():
    # Placeholder function for checkpointing logic
    print("Starting checkpointing process...")
    # TODO: Implement actual checkpointing logic here
    # For example:
    # - Save model state
    # - Save optimizer state
    # - Save current epoch and iteration numbers
    # - Save any other relevant training state

# Main training loop
num_epochs = 100
final_checkpointing_done = False
for epoch in range(num_epochs):
    # TODO: Replace this with your actual training code
    # For example:
    # - Load a batch of data
    # - Forward pass
    # - Calculate loss
    # - Backward pass
    # - Update model parameters
    
    # Check if we're close to capacity expiration and haven't done final checkpointing
    if not final_checkpointing_done and is_close_to_expiration():
        start_checkpointing()
        final_checkpointing_done = True
    
    # Simulate some training time (remove this in actual implementation)
    time.sleep(1)
print("Training completed.")