Ottimizza i modelli su TPU multi-host con Ray e TorchTPU

Autopilot Standard

Questo tutorial mostra come eseguire workload PyTorch distribuiti e ottimizzare un modello Gemma 4 su sezioni Cloud TPU multi-host in Google Kubernetes Engine (GKE) utilizzando Ray Core, Ray Train, PyTorch Fully Sharded Data Parallel (FSDP2) e TorchTPU.

Questo tutorial è rivolto a ML engineer, amministratori e operatori di piattaforme e specialisti di dati e AI che vogliono orchestrare workload di addestramento PyTorch distribuito su sezioni TPU multi-host in Kubernetes. Per scoprire di più sui ruoli comuni e sulle attività di esempio a cui si fa riferimento nei contenuti di Google Cloud , consulta Ruoli e attività comuni degli utenti GKE.

Prima di iniziare questo tutorial, assicurati di avere familiarità con quanto segue:

• Informazioni sulle modalità operative di GKE
• Informazioni sulle TPU in GKE
• Node pool di sezioni TPU multi-host
• Informazioni su Ray su GKE

Sfondo

Questa sezione descrive le tecnologie chiave utilizzate in questo tutorial.

TPU

Le Tensor Processing Unit (TPU) sono acceleratori progettati da Google per i workload di machine learning. Questo tutorial supporta sia TPU Trillium (v6e) sia Ironwood (TPU7x). In una sezione TPU multi-host, più host TPU comunicano direttamente su una rete Inter-Chip Interconnect (ICI) ad alta velocità, fornendo la memoria ad alta larghezza di banda (HBM) aggregata e la comunicazione collettiva a bassa latenza necessarie per addestrare modelli che superano la capacità di memoria di un singolo chip o host TPU.

TorchTPU

TorchTPU è un backend PyTorch open source per Cloud TPU, sviluppato in collaborazione tra Google e Meta. Con TorchTPU, puoi eseguire i modelli PyTorch esistenti su hardware TPU con modifiche minime al codice e scalare i workload di addestramento e inferenza da un singolo chip a pod di TPU multislice tramite il compilatore Accelerated Linear Algebra (XLA). TorchTPU utilizza un modello di esecuzione con un processo per dispositivo sulla rete TPU ICI e supporta le API PyTorch standard, tra cui l'esecuzione eager, torch.compile, torch.distributed, DeviceMesh, DTensor e Fully Sharded Data Parallel (FSDP2).

KubeRay e Ray su TPU

KubeRay è un operatore Kubernetes per il deployment e la gestione delle applicazioni Ray. L'abilitazione del componente aggiuntivo Ray su GKE esegue il deployment dell'operatore KubeRay gestito e del webhook TPU KubeRay. Quando KubeRay crea pod worker che richiedono risorse google.com/tpu, il webhook configura automaticamente i nomi host dei pod, il servizio headless e le variabili di ambiente TPU necessarie per il coordinamento multi-host.

Ray rileva la topologia TPU su ogni nodo all'avvio e pianifica i workload distribuiti tra le sezioni TPU. Puoi orchestrare i workload TorchTPU distribuiti a due livelli:

• Ray Core fornisce gruppi di posizionamento sensibili alla topologia e pianificazione degli attori per loop di esecuzione distribuiti personalizzati, come il calcolo parallelo dei tensori o le pipeline di apprendimento per rinforzo (RL).
• Ray Train fornisce un addestramento gestito con parallelismo dei dati e dei modelli distribuito, tra cui PyTorch Distributed Data Parallel (DDP), PyTorch FSDP2 e JAX single program, multiple data (SPMD), con checkpoint integrato e recupero automatico degli errori.

Obiettivi

Questo tutorial mostra gli aspetti seguenti:

1. Crea un cluster GKE in modalità Autopilot o Standard su TPU Trillium (v6e) o Ironwood (TPU7x).
2. Crea un'immagine container che sovrappone le librerie di addestramento Ray e Hugging Face all'immagine di base TorchTPU.
3. Esegui il deployment di una risorsa personalizzata RayCluster per eseguire il provisioning dei nodi head e worker TPU.
4. Esegui il calcolo distribuito con parallelismo dei tensori con PyTorch DeviceMesh e DTensor in una sezione TPU utilizzando Ray Core.
5. Esegui l'ottimizzazione di un modello Gemma 4 con PyTorch FSDP2 su una sezione TPU multi-host utilizzando Ray Train.

Costi

In questo tutorial utilizzerai i seguenti componenti fatturabili di Google Cloud:

• GKE
• Cloud TPU
• Artifact Registry
• Cloud Storage

Per generare una stima dei costi in base all'utilizzo previsto, utilizza il Calcolatore prezzi.

Al termine delle attività descritte in questo tutorial, puoi evitare l'addebito di ulteriori costi eliminando le risorse che hai creato. Per ulteriori informazioni, consulta Esegui la pulizia.

Prima di iniziare

Prima di iniziare, assicurati di aver eseguito le seguenti attività:

• Abilita l'API Artifact Registry, l'API Cloud Storage e l'API Google Kubernetes Engine.
Abilita le API
• Per utilizzare Google Cloud CLI per questa attività, installala e poi inizializza gcloud CLI. Se hai già installato gcloud CLI, scarica l'ultima versione eseguendo il comando gcloud components update. Le versioni precedenti di gcloud CLI potrebbero non supportare l'esecuzione dei comandi in questo documento.
  Nota: per le installazioni esistenti di gcloud CLI, assicurati di impostare la proprietà compute/region. Se utilizzi principalmente cluster zonali, imposta invece compute/zone. Se imposti una località predefinita, puoi evitare errori in gcloud CLI come il seguente: One of [--zone, --region] must be supplied: Please specify location. Potresti dover specificare la posizione in determinati comandi se la posizione del cluster è diversa da quella predefinita che hai impostato.

• Verifica che il tuo progetto disponga di una quota sufficiente per la capacità TPU Trillium (v6e) o Ironwood (TPU7x) nella zona selezionata. Per maggiori informazioni, consulta Quote di Cloud TPU.
• Assicurati che il tuo account disponga dei seguenti ruoli IAM:
  • Kubernetes Engine Admin (roles/container.admin)
  • Artifact Registry Administrator (roles/artifactregistry.admin)
  • Storage Admin (roles/storage.admin)

prepara l'ambiente

In questo tutorial utilizzerai Cloud Shell per gestire le risorse ospitate su Google Cloud. In Cloud Shell è preinstallato il software necessario per questo tutorial, tra cui kubectl, docker, python3 e l'interfaccia a riga di comando gcloud.

Per configurare l'ambiente con Cloud Shell, segui questi passaggi:

1. Nella console Google Cloud , avvia una sessione di Cloud Shell facendo clic su Attiva Cloud Shell nella consoleGoogle Cloud . Questa azione avvia una sessione nel riquadro inferiore della console Google Cloud .

2. Clona il repository kubernetes-engine-samples e passa alla directory di esempio per questo tutorial:

   git clone https://github.com/GoogleCloudPlatform/kubernetes-engine-samples.git
   cd kubernetes-engine-samples/ai-ml/gke-ray/raytrain/torchtpu
   

3. Crea e attiva un ambiente virtuale Python locale con l'interfaccia a riga di comando Ray installata:

   python3 -m venv ~/ray-env
   source ~/ray-env/bin/activate
   pip install "ray[default]==2.58.0"
   

4. Per scaricare i pesi del modello Gemma 4, accetta il contratto di licenza di Google su Hugging Face:

   1. Vai alla pagina del modello Gemma 4 su Hugging Face.
   2. Accedi e accetta i termini di licenza facendo clic su Accetta e accedi al repository.
   3. Nelle impostazioni del tuo account Hugging Face, genera un token di accesso utente con autorizzazioni Read.

5. Imposta le variabili di ambiente per il tuo progetto Google Cloud , il cluster, il bucket Cloud Storage e il token Hugging Face:

   export PROJECT_ID=PROJECT_ID
   export CLUSTER_NAME=CLUSTER_NAME
   export REGION=REGION
   export ZONE=ZONE
   export ARTIFACT_REGISTRY=ARTIFACT_REGISTRY
   export BUCKET_NAME=BUCKET_NAME
   export HF_TOKEN=HF_TOKEN
   export IMAGE_URI=${REGION}-docker.pkg.dev/${PROJECT_ID}/${ARTIFACT_REGISTRY}/ray-torchtpu:latest
   gcloud config set project ${PROJECT_ID}
   

   Sostituisci quanto segue:

   • PROJECT_ID: il tuo ID progetto Google Cloud .
   • CLUSTER_NAME: il nome del tuo cluster GKE, ad esempio ray-torchtpu-cluster.
   • REGION: la regione in cui si trovano il repository Artifact Registry, il bucket Cloud Storage e la capacità della TPU.
   • ZONE: la zona in cui è disponibile la capacità della TPU. Per le regioni e le zone disponibili in base alla versione della TPU, consulta Regioni e zone TPU.
   • ARTIFACT_REGISTRY: il nome del tuo repository Docker Artifact Registry, ad esempio ray-torchtpu.
   • BUCKET_NAME: il nome del bucket Cloud Storage per l'archiviazione dei checkpoint di Ray Train, ad esempio PROJECT_ID-ray-torchtpu.
   • HF_TOKEN: il tuo token di accesso a Hugging Face.

Crea un cluster GKE

Puoi eseguire workload di addestramento TorchTPU distribuito in un cluster GKE Autopilot o Standard con il componente aggiuntivo operatore Ray gestito abilitato.

1. Crea il cluster GKE e configura le credenziali per lo strumento a riga di comando kubectl:

   Autopilot

   1. Crea un cluster GKE Autopilot regionale con il componente aggiuntivo Operatore Ray abilitato:

      gcloud container clusters create-auto ${CLUSTER_NAME} \
          --project=${PROJECT_ID} \
          --region=${REGION} \
          --cluster-version=1.36 \
          --enable-ray-operator
      

      Quando esegui il deployment della risorsa personalizzata RayCluster in un passaggio successivo, GKE Autopilot esegue automaticamente il provisioning del nodo head della CPU e della sezione TPU multi-host che corrispondono alle richieste di risorse dei pod e ai selettori dei nodi.

   2. Configura lo strumento a riga di comando kubectl per comunicare con il tuo cluster:

      gcloud container clusters get-credentials ${CLUSTER_NAME} \
          --project=${PROJECT_ID} \
          --region=${REGION}
      

   Standard

   1. Crea un cluster GKE Standard di zona con un pool di nodi CPU per ospitare il pod head Ray, Image streaming, GKE Dataplane V2 e la federazione delle identità per i carichi di lavoro per GKE abilitati:

      thought Ecco una selezione di frasi di auguri di buon compleanno per un nipote, suddivise per categoria, così potrai scegliere quella più adatta al suo carattere e al vostro rapporto: ### Affettuose e dolci (perfette per un legame stretto) * "Buon compleanno al nipote migliore del mondo! Ricorda che, ovunque ti porterà la vita, io sarò sempre qui a fare il tifo per te. Ti voglio un bene infinito." * "Vedere il ragazzo/uomo che stai diventando è uno dei regali più belli della mia vita. Ti auguro un anno pieno di felicità, amore e sogni realizzati. Buon compleanno!" * "Non importa quanti anni passino, per me sarai sempre quel bambino meraviglioso che mi ha rubato il cuore. Buon compleanno, nipote mio!" * "A un nipote speciale, che porta gioia e luce nella nostra famiglia ogni singolo giorno. Che la vita ti sorrida sempre. Auguri!" ### Divertenti e simpatiche * "Buon compleanno! Ricorda: più invecchi, più diventi saggio… o almeno questo è quello che dicono per farti sentire meglio! Goditi la tua giornata!" * "Buon compleanno al mio nipote preferito! (Shh, non dirlo agli altri, ma sappiamo entrambi che è

      gcloud container clusters create ${CLUSTER_NAME} \
          --project=${PROJECT_ID} \
          --zone=${ZONE} \
          --cluster-version=1.36 \
          --machine-type=n2-standard-8 \
          --num-nodes=1 \
          --addons=RayOperator \
          --enable-image-streaming \
          --enable-dataplane-v2 \
          --workload-pool=${PROJECT_ID}.svc.id.goog
      

      RayOperator

   2. Configura lo strumento a riga di comando kubectl per comunicare con il tuo cluster:

      gcloud container clusters get-credentials ${CLUSTER_NAME} \
          --project=${PROJECT_ID} \
          --zone=${ZONE}
      

   3. Crea un pool di nodi di sezioni TPU multi-host con 16 dispositivi con una topologia 4x4 su quattro host su TPU Trillium (v6e) o una topologia 2x2x2 su due host su Ironwood (TPU7x):

      Trillium (v6e)

      gcloud container node-pools create tpu-multi-host-pool \
          --project=${PROJECT_ID} \
          --cluster=${CLUSTER_NAME} \
          --zone=${ZONE} \
          --machine-type=ct6e-standard-4t \
          --tpu-topology=4x4 \
          --num-nodes=4
      

      Ironwood (TPU7x)

      gcloud container node-pools create tpu-multi-host-pool \
          --project=${PROJECT_ID} \
          --cluster=${CLUSTER_NAME} \
          --zone=${ZONE} \
          --machine-type=tpu7x-standard-4t \
          --tpu-topology=2x2x2 \
          --num-nodes=2
      

      Suggerimento: per utilizzare una prenotazione TPU specifica, aggiungi i flag --reservation-affinity=specific e --reservation=RESERVATION_NAME al comando node-pools create, dove RESERVATION_NAME è il nome della prenotazione TPU. Per utilizzare le VM spot, aggiungi il flag --spot.

2. Crea un bucket Cloud Storage per archiviare i checkpoint Ray Train condivisi tra i nodi TPU multi-host e concedi alla risorsa ServiceAccount Kubernetes default l'accesso al bucket:

   gcloud storage buckets create gs://${BUCKET_NAME} --location=${REGION}
   
   PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
   gcloud storage buckets add-iam-policy-binding gs://${BUCKET_NAME} \
       --member="principal://iam.googleapis.com/projects/${PROJECT_NUMBER}/locations/global/workloadIdentityPools/${PROJECT_ID}.svc.id.goog/subject/ns/default/sa/default" \
       --role="roles/storage.objectAdmin"
   

3. Crea un secret Kubernetes con il tuo token di accesso Hugging Face in modo che i pod Ray possano scaricare i pesi di Gemma 4 con accesso limitato:

   kubectl create secret generic hf-secret \
       --from-literal=hf_api_token=${HF_TOKEN}
   

Crea l'immagine container

Crea un'immagine container che installi Ray e le librerie di addestramento Hugging Face sull'immagine di base TorchTPU predefinita, che include TorchTPU e la libreria libtpu.

1. Crea un repository Docker di Artifact Registry e configura l'autenticazione Docker:

   gcloud artifacts repositories create ${ARTIFACT_REGISTRY} \
       --repository-format=docker \
       --location=${REGION}
   
   gcloud auth configure-docker ${REGION}-docker.pkg.dev,us-docker.pkg.dev
   

2. Esamina il file Dockerfile nella directory di esempio:

   FROM us-docker.pkg.dev/ml-oss-artifacts-transient/torch-tpu-docker-container/torch-tpu:nightly
   
   # Install wget for KubeRay health probes, Ray (Core, Dashboard, and Train), and Hugging Face libraries for fine-tuning
   RUN apt-get update && apt-get install -y --no-install-recommends wget \
       && rm -rf /var/lib/apt/lists/* \
       && pip install --no-cache-dir \
       "ray[default,train]==2.58.0" \
       "transformers>=5.5.0" \
       datasets

   L'addestramento a sezione singola con la classe Ray TorchTrainer e la pianificazione per chiplet su Ironwood (TPU7x) richiedono Ray versione 2.58.0 o successive, mentre l'ottimizzazione di Gemma 4 richiede la libreria transformers versione 5.5.0 o successive.

3. Crea l'immagine ed eseguine il push in Artifact Registry:

   docker build -t ${IMAGE_URI} .
   docker push ${IMAGE_URI}
   

Esegui il deployment della risorsa personalizzata RayCluster

Esegui il deployment di una risorsa personalizzata RayCluster che esegue il provisioning di un pod head della CPU, di un volume /dev/shm in memoria per l'archivio di oggetti della memoria condivisa di Ray e di un gruppo di worker TPU multi-host.

1. Esamina il manifest RayCluster per la tua versione di TPU nel repository di esempio:

   Trillium (v6e)

   apiVersion: ray.io/v1
   kind: RayCluster
   metadata:
     name: ray-torchtpu-cluster
   spec:
     headGroupSpec:
       rayStartParams:
         num-cpus: "0"
       template:
         spec:
           containers:
           - name: ray-head
             image: ${IMAGE_URI}
             env:
             - name: HF_TOKEN
               valueFrom:
                 secretKeyRef:
                   name: hf-secret
                   key: hf_api_token
             ports:
             - containerPort: 6379
               name: gcs
             - containerPort: 8265
               name: dashboard
             - containerPort: 10001
               name: client
             resources:
               limits:
                 cpu: "4"
                 memory: "16Gi"
               requests:
                 cpu: "4"
                 memory: "16Gi"
     workerGroupSpecs:
     - groupName: v6e-16
       replicas: 1
       numOfHosts: 4
       template:
         spec:
           containers:
           - name: ray-worker
             image: ${IMAGE_URI}
             env:
             - name: HF_TOKEN
               valueFrom:
                 secretKeyRef:
                   name: hf-secret
                   key: hf_api_token
             resources:
               limits:
                 cpu: "64"
                 memory: "200Gi"
                 google.com/tpu: "4"
               requests:
                 cpu: "64"
                 memory: "200Gi"
                 google.com/tpu: "4"
             volumeMounts:
             - mountPath: /dev/shm
               name: dshm
           volumes:
           - name: dshm
             emptyDir:
               medium: Memory
               sizeLimit: 100Gi
           nodeSelector:
             cloud.google.com/gke-tpu-accelerator: tpu-v6e-slice
             cloud.google.com/gke-tpu-topology: 4x4

   Ironwood (TPU7x)

   apiVersion: ray.io/v1
   kind: RayCluster
   metadata:
     name: ray-torchtpu-cluster
   spec:
     headGroupSpec:
       rayStartParams:
         num-cpus: "0"
       template:
         spec:
           containers:
           - name: ray-head
             image: ${IMAGE_URI}
             env:
             - name: HF_TOKEN
               valueFrom:
                 secretKeyRef:
                   name: hf-secret
                   key: hf_api_token
             - name: RAY_TPU_RESOURCE_PER_CHIP
               value: "2"
             ports:
             - containerPort: 6379
               name: gcs
             - containerPort: 8265
               name: dashboard
             - containerPort: 10001
               name: client
             resources:
               limits:
                 cpu: "4"
                 memory: "16Gi"
               requests:
                 cpu: "4"
                 memory: "16Gi"
     workerGroupSpecs:
     - groupName: tpu7x-16
       replicas: 1
       numOfHosts: 2
       resources:
         # Advertises 8 logical TPU cores (2 chiplets per physical chip) per host.
         TPU: "8"
       template:
         spec:
           containers:
           - name: ray-worker
             image: ${IMAGE_URI}
             env:
             - name: HF_TOKEN
               valueFrom:
                 secretKeyRef:
                   name: hf-secret
                   key: hf_api_token
             - name: RAY_TPU_RESOURCE_PER_CHIP
               value: "2"
             resources:
               limits:
                 cpu: "64"
                 memory: "250Gi"
                 google.com/tpu: "4"
               requests:
                 cpu: "64"
                 memory: "250Gi"
                 google.com/tpu: "4"
             volumeMounts:
             - mountPath: /dev/shm
               name: dshm
           volumes:
           - name: dshm
             emptyDir:
               medium: Memory
               sizeLimit: 128Gi
           nodeSelector:
             cloud.google.com/gke-tpu-accelerator: tpu7x
             cloud.google.com/gke-tpu-topology: 2x2x2

   Su Ironwood (TPU7x), ogni chip fisico contiene due chiplet, fornendo otto core TPU logici per host a quattro chip. Il manifest imposta il valore del campo TPU su "8" nella sezione resources e imposta la variabile di ambiente RAY_TPU_RESOURCE_PER_CHIP su "2" per annunciare e pianificare otto risorse TPU logiche per host.

2. Applica il manifest per creare la risorsa personalizzata RayCluster:

   Trillium (v6e)

   envsubst < ray-cluster.tpu-v6e-multihost.yaml | kubectl apply -f -
   

   Ironwood (TPU7x)

   envsubst < ray-cluster.tpu-v7x-multihost.yaml | kubectl apply -f -
   

3. Attendi finché il pod head di RayCluster e tutti i pod worker TPU non raggiungono lo stato Running:

   kubectl get pods -l ray.io/cluster=ray-torchtpu-cluster
   

   L'output è simile al seguente:

   Trillium (v6e)

   NAME                                         READY   STATUS    RESTARTS   AGE
   ray-torchtpu-cluster-head-pdpk6              1/1     Running   0          2m30s
   ray-torchtpu-cluster-v6e-16-worker-brwpc     1/1     Running   0          2m30s
   ray-torchtpu-cluster-v6e-16-worker-dx4w7     1/1     Running   0          2m30s
   ray-torchtpu-cluster-v6e-16-worker-h4d4w     1/1     Running   0          2m30s
   ray-torchtpu-cluster-v6e-16-worker-t2g24     1/1     Running   0          2m30s
   

   Ironwood (TPU7x)

   NAME                                         READY   STATUS    RESTARTS   AGE
   ray-torchtpu-cluster-head-k9x2m              1/1     Running   0          2m30s
   ray-torchtpu-cluster-tpu7x-16-worker-7cwpq   1/1     Running   0          2m30s
   ray-torchtpu-cluster-tpu7x-16-worker-m98r9   1/1     Running   0          2m30s
   

4. Ispeziona le variabili di ambiente su uno dei pod worker per verificare che il webhook TPU KubeRay abbia inserito le variabili di coordinamento TPU multi-host e di topologia TorchTPU:

   kubectl get pods -l ray.io/node-type=worker \
       -o=jsonpath='{range .items[0].spec.containers[0].env[*]}{.name}={.value}{"\n"}{end}' \
       | grep -E "TPU_WORKER_ID|TPU_WORKER_HOSTNAMES|TPU_NAME|TORCH_TPU_TOPOLOGY|TORCH_TPU_SLICEBUILDER_ADDRESSES"
   

   L'output è simile al seguente:

   Trillium (v6e)

   TPU_WORKER_HOSTNAMES=v6e-16-0-0.ray-torchtpu-cluster-headless,v6e-16-0-1.ray-torchtpu-cluster-headless,v6e-16-0-2.ray-torchtpu-cluster-headless,v6e-16-0-3.ray-torchtpu-cluster-headless
   TORCH_TPU_TOPOLOGY=4,4,1
   TORCH_TPU_SLICEBUILDER_ADDRESSES=v6e-16-0-0.ray-torchtpu-cluster-headless:8471,v6e-16-0-0.ray-torchtpu-cluster-headless:8472,v6e-16-0-0.ray-torchtpu-cluster-headless:8473,v6e-16-0-0.ray-torchtpu-cluster-headless:8474,v6e-16-0-1.ray-torchtpu-cluster-headless:8471,v6e-16-0-1.ray-torchtpu-cluster-headless:8472,v6e-16-0-1.ray-torchtpu-cluster-headless:8473,v6e-16-0-1.ray-torchtpu-cluster-headless:8474,v6e-16-0-2.ray-torchtpu-cluster-headless:8471,v6e-16-0-2.ray-torchtpu-cluster-headless:8472,v6e-16-0-2.ray-torchtpu-cluster-headless:8473,v6e-16-0-2.ray-torchtpu-cluster-headless:8474,v6e-16-0-3.ray-torchtpu-cluster-headless:8471,v6e-16-0-3.ray-torchtpu-cluster-headless:8472,v6e-16-0-3.ray-torchtpu-cluster-headless:8473,v6e-16-0-3.ray-torchtpu-cluster-headless:8474
   TPU_WORKER_ID=3
   TPU_NAME=v6e-16-0
   

   Ironwood (TPU7x)

   TPU_WORKER_HOSTNAMES=tpu7x-16-0-0.ray-torchtpu-cluster-headless,tpu7x-16-0-1.ray-torchtpu-cluster-headless
   TORCH_TPU_TOPOLOGY=2,2,2,2
   TORCH_TPU_SLICEBUILDER_ADDRESSES=tpu7x-16-0-0.ray-torchtpu-cluster-headless:8471,tpu7x-16-0-0.ray-torchtpu-cluster-headless:8472,tpu7x-16-0-0.ray-torchtpu-cluster-headless:8473,tpu7x-16-0-0.ray-torchtpu-cluster-headless:8474,tpu7x-16-0-0.ray-torchtpu-cluster-headless:8475,tpu7x-16-0-0.ray-torchtpu-cluster-headless:8476,tpu7x-16-0-0.ray-torchtpu-cluster-headless:8477,tpu7x-16-0-0.ray-torchtpu-cluster-headless:8478,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8471,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8472,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8473,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8474,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8475,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8476,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8477,tpu7x-16-0-1.ray-torchtpu-cluster-headless:8478
   TPU_WORKER_ID=1
   TPU_NAME=tpu7x-16-0
   

5. Stabilisci una sessione di port forwarding al servizio head Ray:

   kubectl port-forward svc/ray-torchtpu-cluster-head-svc 8265:8265 >/dev/null 2>&1 &
   

   Puoi ispezionare i nodi worker TPU e le relative etichette ray.io/tpu-* rilevate automaticamente nella scheda Cluster della dashboard Ray.

Esegui il parallelismo dei tensori distribuito con Ray Core

Ray Core fornisce la funzione slice_placement_group nel modulo ray.util.tpu per prenotare una sezione TPU multi-host intatta e la classe PlacementGroupSchedulingStrategy per pianificare un attore Ray per dispositivo TPU logico nella sezione.

1. Esamina lo script ray_core_collectives.py nel repository di esempio:

   import argparse
   import os
   
   import ray
   from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
   from ray.util.tpu import (
       get_tpu_version_from_type,
       slice_placement_group,
   )
   import torch
   import torch.distributed as dist
   from torch.distributed.device_mesh import init_device_mesh
   from torch.distributed.tensor.parallel import (
       ColwiseParallel,
       RowwiseParallel,
       parallelize_module,
   )
   import torch.nn as nn
   import torch_tpu  # noqa: F401
   
   RANDOM_SEED = 42
   
   
   class FeedForward(nn.Module):
       """Two-layer feed-forward block for tensor-parallel sharding across a TPU slice."""
   
       def __init__(self, hidden_dim: int):
           super().__init__()
           self.w1 = nn.Linear(hidden_dim, 4 * hidden_dim, bias=False)
           self.act = nn.SiLU()
           self.w2 = nn.Linear(4 * hidden_dim, hidden_dim, bias=False)
   
       def forward(self, x: torch.Tensor) -> torch.Tensor:
           return x + self.w2(self.act(self.w1(x)))
   
   
   @ray.remote
   class TorchTPUSliceWorker:
       """Ray actor bound to a single logical TPU device within a slice placement group."""
   
       def __init__(self, devices_per_host: int):
           self.node_ip = ray.util.get_node_ip_address()
           hosts = os.environ["TPU_WORKER_HOSTNAMES"].split(",")
           self.worker_id = int(os.environ.get("TPU_WORKER_ID", "0"))
           self.local_rank = int(
               ray.get_runtime_context().get_accelerator_ids()["TPU"][0]
           )
           self.rank = self.worker_id * devices_per_host + self.local_rank
           # Set per-process PyTorch env:// variables from the webhook's host-level
           # TPU metadata and Ray's assigned local device ID (TorchTrainer sets these
           # automatically in the Ray Train section below).
           os.environ.update(
               {
                   "MASTER_ADDR": hosts[0],
                   "MASTER_PORT": "29500",
                   "WORLD_SIZE": str(len(hosts) * devices_per_host),
                   "RANK": str(self.rank),
                   "LOCAL_RANK": str(self.local_rank),
                   "LOCAL_WORLD_SIZE": str(devices_per_host),
               }
           )
   
       def run_tensor_parallel_step(
           self,
           batch_size: int = 64,
           hidden_dim: int = 128,
       ) -> dict:
           dist.init_process_group(backend="tpu_dist")
           world_size = dist.get_world_size()
           device = torch.device("tpu")
   
           # 1. Compute single-device reference output on CPU with identical seed.
           torch.manual_seed(RANDOM_SEED)
           ref_model = FeedForward(hidden_dim).eval()
           inputs_cpu = torch.randn(batch_size, hidden_dim)
           with torch.no_grad():
               ref_output = ref_model(inputs_cpu)
   
           # 2. Shard the model across all TPU devices in the slice with DTensor TP.
           torch.manual_seed(RANDOM_SEED)
           tp_model = FeedForward(hidden_dim).to(device).eval()
           mesh = init_device_mesh("tpu", (world_size,))
           tp_model = parallelize_module(
               tp_model,
               mesh,
               {"w1": ColwiseParallel(), "w2": RowwiseParallel()},
           )
   
           with torch.no_grad():
               tp_output = tp_model(inputs_cpu.to(device)).cpu()
   
           torch.testing.assert_close(tp_output, ref_output, rtol=1e-3, atol=1e-2)
           max_abs_diff = float((tp_output - ref_output).abs().max().item())
   
           # 3. Verify direct ICI all_reduce across all ranks in the slice.
           rank_tensor = torch.full(
               (4,), float(self.rank + 1), dtype=torch.float32, device=device
           )
           dist.all_reduce(rank_tensor, op=dist.ReduceOp.SUM)
           all_reduce_sum = rank_tensor.cpu().tolist()
   
           dist.destroy_process_group()
           return {
               "rank": self.rank,
               "local_rank": self.local_rank,
               "worker_id": self.worker_id,
               "node_ip": self.node_ip,
               "all_reduce_sum": all_reduce_sum,
               "max_abs_diff": max_abs_diff,
           }
   
   
   def parse_args() -> argparse.Namespace:
       parser = argparse.ArgumentParser(
           description="Run distributed TorchTPU tensor parallelism across a Ray TPU slice."
       )
       parser.add_argument(
           "--topology",
           type=str,
           default="4x4",
           help="TPU physical topology.",
       )
       parser.add_argument(
           "--accelerator-type",
           type=str,
           default="TPU-V6E",
           help="Ray TPU accelerator type (TPU-V6E, TPU-V7X, etc.).",
       )
       return parser.parse_args()
   
   
   def main() -> None:
       args = parse_args()
       ray.init()
   
       slice_handle = slice_placement_group(
           topology=args.topology,
           accelerator_version=get_tpu_version_from_type(args.accelerator_type),
           num_slices=1,
           resources_per_bundle={"TPU": 1},
       )
       ray.get(slice_handle.placement_group.ready())
   
       devices_per_host = slice_handle.num_bundles // slice_handle.num_hosts
       workers = [
           TorchTPUSliceWorker.options(
               resources={"TPU": 1},
               scheduling_strategy=PlacementGroupSchedulingStrategy(
                   placement_group=slice_handle.placement_group,
                   placement_group_bundle_index=bundle_idx,
               ),
           ).remote(devices_per_host)
           for bundle_idx in range(slice_handle.num_bundles)
       ]
       results = sorted(
           ray.get([w.run_tensor_parallel_step.remote() for w in workers]),
           key=lambda r: r["rank"],
       )
   
       expected_sum = sum(range(1, slice_handle.num_bundles + 1))
       for r in results:
           print(
               f"Rank {r['rank']:2d} (worker_id={r['worker_id']}, "
               f"local_rank={r['local_rank']}, ip={r['node_ip']}): "
               f"all_reduce_sum={r['all_reduce_sum']}"
           )
           assert r["all_reduce_sum"] == [float(expected_sum)] * 4
   
       print(
           f"All {len(results)} TorchTPU ranks verified DTensor TP parity "
           f"(max_abs_diff={results[0]['max_abs_diff']:.6f}) and "
           f"all_reduce sum = {expected_sum:.1f}"
       )
       for w in workers:
           ray.kill(w)
       slice_handle.shutdown()
   
   
   if __name__ == "__main__":
       main()

   Lo script esegue il parallelismo dei tensori distribuito nella sezione TPU in tre passaggi:

   1. La funzione slice_placement_group prenota una sezione TPU di 16 dispositivi e pianifica un attore TorchTPUSliceWorker su ogni dispositivo TPU.
   2. Ogni attore calcola il proprio ranking globale dall'indice dell'host e dall'ID dispositivo locale, quindi inizializza il gruppo di processi tpu_dist.
   3. Le funzioni init_device_mesh e parallelize_module suddividono in shard i livelli feed-forward su tutti i 16 dispositivi TPU, verificano l'output rispetto a un riferimento CPU su un singolo dispositivo ed eseguono un'operazione collettiva all_reduce su ICI.

2. Invia il job Ray Core per la tua versione di TPU:

   Trillium (v6e)

   ray job submit \
       --address http://localhost:8265 \
       --working-dir . \
       --runtime-env-json '{"excludes": ["ray-env", ".git", "*.whl"]}' \
       -- python3 ray_core_collectives.py --topology=4x4 --accelerator-type=TPU-V6E
   

   Ironwood (TPU7x)

   ray job submit \
       --address http://localhost:8265 \
       --working-dir . \
       --runtime-env-json '{"excludes": ["ray-env", ".git", "*.whl"]}' \
       -- python3 ray_core_collectives.py --topology=2x2x2 --accelerator-type=TPU-V7X
   

   L'output è simile al seguente:

   Trillium (v6e)

   Rank  0 (worker_id=0, local_rank=0, ip=10.72.3.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  1 (worker_id=0, local_rank=1, ip=10.72.3.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  2 (worker_id=0, local_rank=2, ip=10.72.3.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  3 (worker_id=0, local_rank=3, ip=10.72.3.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  4 (worker_id=1, local_rank=0, ip=10.72.14.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  5 (worker_id=1, local_rank=1, ip=10.72.14.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  6 (worker_id=1, local_rank=2, ip=10.72.14.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  7 (worker_id=1, local_rank=3, ip=10.72.14.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  8 (worker_id=2, local_rank=0, ip=10.72.7.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  9 (worker_id=2, local_rank=1, ip=10.72.7.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 10 (worker_id=2, local_rank=2, ip=10.72.7.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 11 (worker_id=2, local_rank=3, ip=10.72.7.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 12 (worker_id=3, local_rank=0, ip=10.72.6.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 13 (worker_id=3, local_rank=1, ip=10.72.6.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 14 (worker_id=3, local_rank=2, ip=10.72.6.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 15 (worker_id=3, local_rank=3, ip=10.72.6.20): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   All 16 TorchTPU ranks verified DTensor TP parity (max_abs_diff=0.002470) and all_reduce sum = 136.0
   ------------------------------------------
   Job 'raysubmit_HFTkKrjSnfza3ZFD' succeeded
   ------------------------------------------
   

   Ironwood (TPU7x)

   Rank  0 (worker_id=0, local_rank=0, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  1 (worker_id=0, local_rank=1, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  2 (worker_id=0, local_rank=2, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  3 (worker_id=0, local_rank=3, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  4 (worker_id=0, local_rank=4, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  5 (worker_id=0, local_rank=5, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  6 (worker_id=0, local_rank=6, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  7 (worker_id=0, local_rank=7, ip=172.16.14.6): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  8 (worker_id=1, local_rank=0, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank  9 (worker_id=1, local_rank=1, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 10 (worker_id=1, local_rank=2, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 11 (worker_id=1, local_rank=3, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 12 (worker_id=1, local_rank=4, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 13 (worker_id=1, local_rank=5, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 14 (worker_id=1, local_rank=6, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   Rank 15 (worker_id=1, local_rank=7, ip=172.16.15.7): all_reduce_sum=[136.0, 136.0, 136.0, 136.0]
   All 16 TorchTPU ranks verified DTensor TP parity (max_abs_diff=0.002470) and all_reduce sum = 136.0
   ------------------------------------------
   Job 'raysubmit_MrJbAe3hvARBcBN2' succeeded
   ------------------------------------------
   

Ottimizza un modello Gemma 4 con Ray Train e PyTorch FSDP2

La classe Ray Train TorchTrainer automatizza la prenotazione delle sezioni TPU, il posizionamento dei worker e l'inizializzazione di torch.distributed in una sezione TPU multi-host.

1. Esamina lo script ray_train_llm_finetune.py nel repository di esempio, che ottimizza il modello google/gemma-4-E2B-it sul set di dati di istruzioni databricks/databricks-dolly-15k:

   import argparse
   import os
   import tempfile
   import time
   
   from datasets import load_dataset
   import ray
   import ray.train
   from ray.train import (
       Checkpoint,
       CheckpointConfig,
       RunConfig,
       ScalingConfig,
   )
   from ray.train.torch import TorchConfig, TorchTrainer
   import torch
   from torch.distributed import ReduceOp, all_reduce
   from torch.distributed.checkpoint.state_dict import (
       StateDictOptions,
       get_model_state_dict,
   )
   from torch.distributed.device_mesh import init_device_mesh
   from torch.distributed.fsdp import MixedPrecisionPolicy, fully_shard
   from torch.utils.data import DataLoader
   import torch_tpu  # noqa: F401
   from transformers import (
       AutoModelForCausalLM,
       AutoTokenizer,
       DataCollatorForLanguageModeling,
   )
   
   
   def create_dataloader(
       tokenizer: AutoTokenizer,
       batch_size: int,
       dataset_split: str = "train[:1024]",
       max_seq_len: int = 512,
   ) -> DataLoader:
       """Loads and tokenizes the Dolly 15k dataset for causal LM fine-tuning."""
       dataset = load_dataset(
           "databricks/databricks-dolly-15k", split=dataset_split
       )
       tokenized = dataset.map(
           lambda x: tokenizer(
               f"Instruction: {x['instruction']}\nResponse: {x['response']}",
               truncation=True,
               max_length=max_seq_len,
               padding="max_length",
           ),
           remove_columns=dataset.column_names,
       )
       dataloader = DataLoader(
           tokenized,
           batch_size=batch_size,
           shuffle=True,
           drop_last=True,
           collate_fn=DataCollatorForLanguageModeling(
               tokenizer=tokenizer, mlm=False
           ),
       )
       return ray.train.torch.prepare_data_loader(dataloader)
   
   
   def train_func(config: dict) -> None:
       """Per-worker distributed FSDP2 training loop executed on each logical TPU core."""
       ctx = ray.train.get_context()
       world_rank = ctx.get_world_rank()
       world_size = ctx.get_world_size()
       device = ray.train.torch.get_device()
   
       model_id = config["model_id"]
       dataset_split = config["dataset_split"]
       batch_size = config["batch_size_per_device"]
       num_epochs = config["num_epochs"]
       max_seq_len = 512
   
       tokenizer = AutoTokenizer.from_pretrained(model_id)
       dataloader = create_dataloader(
           tokenizer=tokenizer,
           batch_size=batch_size,
           dataset_split=dataset_split,
           max_seq_len=max_seq_len,
       )
   
       torch.manual_seed(42)
       model = AutoModelForCausalLM.from_pretrained(
           model_id,
           dtype=torch.bfloat16,
           attn_implementation="eager",
           vision_config=None,
           audio_config=None,
       ).to(device)
       model.config.get_text_config().use_cache = False
   
       mesh = init_device_mesh("tpu", (world_size,))
       mp_policy = MixedPrecisionPolicy(
           param_dtype=torch.bfloat16,
           reduce_dtype=torch.float32,
       )
       text_model = model.model.language_model
       text_model.embed_tokens_per_layer.requires_grad_(False)
       for layer in text_model.layers:
           fully_shard(layer.mlp, mesh=mesh, mp_policy=mp_policy)
       fully_shard(
           model,
           mesh=mesh,
           mp_policy=mp_policy,
           ignored_params=set(text_model.embed_tokens_per_layer.parameters()),
       )
   
       optimizer = torch.optim.AdamW(
           model.parameters(), lr=2e-5, weight_decay=0.01
       )
       global_batch_size = batch_size * world_size
   
       for epoch in range(num_epochs):
           dataloader.sampler.set_epoch(epoch)
           model.train()
           epoch_start = time.perf_counter()
           running_loss = torch.zeros(1, device=device, dtype=torch.float32)
   
           for batch in dataloader:
               optimizer.zero_grad()
               outputs = model(
                   input_ids=batch["input_ids"],
                   labels=batch["labels"],
               )
               outputs.loss.backward()
               optimizer.step()
               running_loss += outputs.loss.detach().float()
   
           all_reduce(running_loss, op=ReduceOp.AVG)
           avg_loss = running_loss.item() / len(dataloader)
           epoch_time = time.perf_counter() - epoch_start
           tokens_per_sec = (
               len(dataloader) * global_batch_size * max_seq_len
           ) / epoch_time
   
           metrics = {
               "epoch": epoch + 1,
               "loss": round(avg_loss, 4),
               "epoch_time_s": round(epoch_time, 2),
               "tokens_per_s": round(tokens_per_sec, 1),
           }
           if world_rank == 0:
               print(
                   f"Epoch {epoch + 1}/{num_epochs} | loss={metrics['loss']:.4f} | "
                   f"epoch_time={metrics['epoch_time_s']:.2f}s | "
                   f"tokens/s={metrics['tokens_per_s']:.1f}"
               )
   
           full_state_dict = get_model_state_dict(
               model,
               options=StateDictOptions(
                   full_state_dict=True,
                   cpu_offload=True,
                   ignore_frozen_params=True,
               ),
           )
           if world_rank == 0:
               with tempfile.TemporaryDirectory() as tmpdir:
                   torch.save(full_state_dict, os.path.join(tmpdir, "model.pt"))
                   ray.train.report(
                       metrics, checkpoint=Checkpoint.from_directory(tmpdir)
                   )
           else:
               ray.train.report(metrics)
   
   
   def parse_args() -> argparse.Namespace:
       parser = argparse.ArgumentParser(
           description="Fine-tune a Hugging Face CausalLM on TPUs with Ray Train and TorchTPU FSDP2."
       )
       parser.add_argument(
           "--model-id",
           type=str,
           default="google/gemma-4-E2B-it",
           help="Hugging Face model ID to fine-tune.",
       )
       parser.add_argument(
           "--topology",
           type=str,
           default="4x4",
           help="TPU slice topology.",
       )
       parser.add_argument(
           "--accelerator-type",
           type=str,
           default="TPU-V6E",
           help="Ray TPU accelerator type (TPU-V6E, TPU-V7X, etc.).",
       )
       parser.add_argument(
           "--num-workers",
           type=int,
           default=16,
           help="Total TPU worker processes across the slice.",
       )
       parser.add_argument(
           "--dataset-split",
           type=str,
           default="train[:1024]",
           help="Hugging Face dataset split to load.",
       )
       parser.add_argument(
           "--num-epochs",
           type=int,
           default=3,
           help="Number of fine-tuning epochs.",
       )
       parser.add_argument(
           "--batch-size-per-device",
           type=int,
           default=4,
           help="Per-device training batch size.",
       )
       parser.add_argument(
           "--storage-path",
           type=str,
           default=os.environ.get("STORAGE_PATH"),
           required="STORAGE_PATH" not in os.environ,
           help="Cloud Storage or shared file system path for checkpoints.",
       )
       return parser.parse_args()
   
   
   def main() -> None:
       args = parse_args()
       ray.init()
       run_name = f"torchtpu_gemma4_fsdp2_sft_{int(time.time())}"
   
       print(
           f"Starting Ray Train FSDP2 SFT job: model={args.model_id}, "
           f"topology={args.topology}, accelerator_type={args.accelerator_type}, "
           f"num_workers={args.num_workers}"
       )
   
       trainer = TorchTrainer(
           train_loop_per_worker=train_func,
           train_loop_config={
               "model_id": args.model_id,
               "dataset_split": args.dataset_split,
               "num_epochs": args.num_epochs,
               "batch_size_per_device": args.batch_size_per_device,
           },
           torch_config=TorchConfig(backend="tpu_dist"),
           scaling_config=ScalingConfig(
               use_tpu=True,
               num_workers=args.num_workers,
               topology=args.topology,
               accelerator_type=args.accelerator_type,
               resources_per_worker={"TPU": 1},
           ),
           run_config=RunConfig(
               name=run_name,
               storage_path=args.storage_path,
               checkpoint_config=CheckpointConfig(
                   num_to_keep=1,
                   checkpoint_score_attribute="loss",
                   checkpoint_score_order="min",
               ),
           ),
       )
   
       result = trainer.fit()
       print(f"Training completed! Final metrics: {result.metrics}")
       print(f"Saved checkpoint: {result.checkpoint}")
   
   
   if __name__ == "__main__":
       main()

   Lo script configura la classe TorchTrainer con i seguenti componenti:

   • La classe ScalingConfig configura la topologia dei worker distribuiti per il job di addestramento. L'impostazione dei parametri use_tpu=True, num_workers=16, topology e accelerator_type prenota una sezione TPU di 16 dispositivi e pianifica un worker per core TPU.
   • La classe TorchConfig configura il gruppo di processi torch.distributed tra i worker. L'impostazione del parametro backend="tpu_dist" connette tutti i 16 worker sulla rete TPU ICI prima dell'esecuzione della funzione train_func.
   • La funzione create_dataloader scarica e tokenizza il set di dati di istruzioni, raggruppa le sequenze di token con un oggetto PyTorch DataLoader e chiama la funzione ray.train.torch.prepare_data_loader in modo che ogni worker riceva uno shard disgiunto di ogni batch globale sul dispositivo TPU assegnato.
   • La funzione train_func carica il modello con precisione bfloat16 e chiama la funzione PyTorch FSDP2 fully_shard su ogni blocco feed-forward e sul modello root per eseguire lo sharding di parametri, gradienti e stati dell'ottimizzatore su tutti i 16 core TPU.
   • La classe Checkpoint e la funzione ray.train.report raccolgono il dizionario completo dello stato del modello nel rank 0 alla fine di ogni epoca e caricano il checkpoint nel bucket Cloud Storage.

2. Invia il job di ottimizzazione TorchTrainer FSDP2 per la tua versione di TPU:

   Trillium (v6e)

   ray job submit \
       --address http://localhost:8265 \
       --working-dir . \
       --runtime-env-json '{"excludes": ["ray-env", ".git", "*.whl"]}' \
       -- python3 ray_train_llm_finetune.py \
           --topology=4x4 \
           --accelerator-type=TPU-V6E \
           --storage-path=gs://${BUCKET_NAME}/ray_train_torchtpu
   

   Ironwood (TPU7x)

   ray job submit \
       --address http://localhost:8265 \
       --working-dir . \
       --runtime-env-json '{"excludes": ["ray-env", ".git", "*.whl"]}' \
       -- python3 ray_train_llm_finetune.py \
           --topology=2x2x2 \
           --accelerator-type=TPU-V7X \
           --storage-path=gs://${BUCKET_NAME}/ray_train_torchtpu
   

   Suggerimento: per impostazione predefinita, lo script esegue l'addestramento su un sottoinsieme di 1024 esempi per tre epoche,in modo che il job venga completato in pochi minuti. Per eseguire l'addestramento su tutti i 15.011 esempi o modificare il numero di epoche, passa il flag --dataset-split=train o --num-epochs allo script ray_train_llm_finetune.py.

   L'output è simile al seguente:

   Trillium (v6e)

   Starting Ray Train FSDP2 SFT job: model=google/gemma-4-E2B-it, topology=4x4, accelerator_type=TPU-V6E, num_workers=16
   (TrainController pid=5407) Attempting to start training worker group of size 16 with the following resources: [{'TPU': 1, 'accelerator_type:TPU-V6E': 0.001}] * 16
   (RayTrainWorker pid=19733, ip=10.72.3.20) Setting up process group for: env:// [rank=0, world_size=16]
   (RayTrainWorker pid=19733, ip=10.72.3.20) Epoch 1/3 | loss=3.1402 | epoch_time=28.17s | tokens/s=18613.5
   (RayTrainWorker pid=19733, ip=10.72.3.20) Checkpoint successfully created at: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791864427/checkpoint_2026-10-13_04-08-10.762175)
   (RayTrainWorker pid=19733, ip=10.72.3.20) Epoch 2/3 | loss=2.3318 | epoch_time=26.07s | tokens/s=20112.8
   (RayTrainWorker pid=19733, ip=10.72.3.20) Checkpoint successfully created at: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791864427/checkpoint_2026-10-13_04-10-02.548088)
   (RayTrainWorker pid=19733, ip=10.72.3.20) Epoch 3/3 | loss=2.2050 | epoch_time=25.52s | tokens/s=20545.2
   (RayTrainWorker pid=19733, ip=10.72.3.20) Checkpoint successfully created at: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791864427/checkpoint_2026-10-13_04-11-48.701853)
   Training completed! Final metrics: {'epoch': 3, 'loss': 2.205, 'epoch_time_s': 25.52, 'tokens_per_s': 20545.2}
   Saved checkpoint: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791864427/checkpoint_2026-10-13_04-11-48.701853)
   ------------------------------------------
   Job 'raysubmit_zbm4pnFZitA1YECA' succeeded
   ------------------------------------------
   

   Ironwood (TPU7x)

   Starting Ray Train FSDP2 SFT job: model=google/gemma-4-E2B-it, topology=2x2x2, accelerator_type=TPU-V7X, num_workers=16
   (TrainController pid=1162) Attempting to start training worker group of size 16 with the following resources: [{'TPU': 1, 'accelerator_type:TPU-V7X': 0.001}] * 16
   (RayTrainWorker pid=257, ip=172.16.26.6) Setting up process group for: env:// [rank=0, world_size=16]
   (RayTrainWorker pid=257, ip=172.16.26.6) Epoch 1/3 | loss=3.1407 | epoch_time=26.28s | tokens/s=19953.3
   (RayTrainWorker pid=257, ip=172.16.26.6) Checkpoint successfully created at: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791867004/checkpoint_2026-10-13_04-52-00.026930)
   (RayTrainWorker pid=257, ip=172.16.26.6) Epoch 2/3 | loss=2.3235 | epoch_time=16.68s | tokens/s=31429.9
   (RayTrainWorker pid=257, ip=172.16.26.6) Checkpoint successfully created at: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791867004/checkpoint_2026-10-13_04-53-33.533341)
   (RayTrainWorker pid=257, ip=172.16.26.6) Epoch 3/3 | loss=2.1964 | epoch_time=16.98s | tokens/s=30884.6
   (RayTrainWorker pid=257, ip=172.16.26.6) Checkpoint successfully created at: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791867004/checkpoint_2026-10-13_04-55-01.508219)
   Training completed! Final metrics: {'epoch': 3, 'loss': 2.1964, 'epoch_time_s': 16.98, 'tokens_per_s': 30884.6}
   Saved checkpoint: Checkpoint(filesystem=gcs, path=my-project-ray-torchtpu/ray_train_torchtpu/torchtpu_gemma4_fsdp2_sft_1791867004/checkpoint_2026-10-13_04-55-01.508219)
   ------------------------------------------
   Job 'raysubmit_zSNnCXhHkwKiSmJh' succeeded
   ------------------------------------------
   

   L'output dell'addestramento mostra quanto segue:

   • L'epoca 1 include la compilazione iniziale del grafico XLA. Le epoche 2 e 3 riutilizzano gli eseguibili memorizzati nella cache di /dev/shm/torch_tpu_cache, raggiungendo oltre 20.000 token al secondo su TPU Trillium (v6e) e 31.000 token al secondo su Ironwood (TPU7x).
   • La perdita di entropia incrociata globale sincronizzata tra tutti i 16 rank dei worker FSDP2 diminuisce da 3,14 nell'epoca 1 a 2,20 nell'epoca 3 nelle tre epoche di ottimizzazione sul set di dati di istruzioni databricks/databricks-dolly-15k.
   • Alla fine di ogni epoca, il rank 0 raccoglie il dizionario completo dello stato del modello e carica il checkpoint nel bucket Cloud Storage, conservando il checkpoint con la perdita più bassa, come configurato nell'oggetto CheckpointConfig(num_to_keep=1, checkpoint_score_attribute="loss", checkpoint_score_order="min").

Operazione riuscita: hai eseguito un calcolo distribuito con parallelismo dei tensori con Ray Core e hai ottimizzato un modello Gemma 4 con PyTorch FSDP2 su TPU in GKE utilizzando Ray Train e TorchTPU.

Osserva le prestazioni del workload

Per monitorare l'utilizzo di TensorCore della TPU, l'utilizzo di HBM e le metriche del cluster Ray in Cloud Monitoring, consulta Raccogli e visualizza log e metriche per i cluster Ray su GKE e Visualizza le metriche della TPU nella dashboard di Ray.

Accedi alla dashboard Ray

Per esaminare lo stato degli attori, dei gruppi di posizionamento e dei job di addestramento Ray nella dashboard Ray integrata, segui questi passaggi:

1. Se il port forwarding al servizio head di Ray non è già in esecuzione, avvialo in background:

   kubectl port-forward svc/ray-torchtpu-cluster-head-svc 8265:8265 >/dev/null 2>&1 &
   

2. In Cloud Shell, fai clic su Anteprima web e seleziona Anteprima sulla porta 8265 oppure apri http://localhost:8265 nel browser locale.

3. Fai clic sulle schede Job, Cluster e Metriche per esaminare i log dei worker, lo stato del gruppo di posizionamento, le etichette dei nodi TPU e il grafico Utilizzo TPU per nodo.

Esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo tutorial, elimina la risorsa personalizzata RayCluster, il cluster GKE, il bucket Cloud Storage e il repository Artifact Registry:

1. Interrompi il processo di port forwarding in background ed elimina la risorsa personalizzata RayCluster e il secret Kubernetes:

   pkill -f "kubectl port-forward" || true
   kubectl delete raycluster ray-torchtpu-cluster
   kubectl delete secret hf-secret --ignore-not-found
   

2. Elimina il cluster GKE:

   Autopilot

   gcloud container clusters delete ${CLUSTER_NAME} \
       --region=${REGION}
   

   Standard

   gcloud container clusters delete ${CLUSTER_NAME} \
       --zone=${ZONE}
   

3. Elimina il bucket Cloud Storage e i relativi contenuti dei checkpoint:

   gcloud storage rm --recursive gs://${BUCKET_NAME}
   

4. Elimina il repository Artifact Registry:

   gcloud artifacts repositories delete ${ARTIFACT_REGISTRY} \
       --location=${REGION}
   

Passaggi successivi

• Scopri di più su PyTorch su Cloud TPU.
• Consulta la documentazione di TorchTPU.
• Scopri come gestire gli LLM su TPU multi-host con Ray e TorchTPU.
• Scopri come ottimizzare l'addestramento dell'AI su TPU con DWS e Kueue.
• Leggi Informazioni sulle TPU in GKE.
• Consulta la documentazione di Ray Train.