Gestire i file tra le esecuzioni

Un job di un workflow gestito funziona come un job batch su un nodo di calcolo HPC. Il disco locale del job è spazio temporaneo. Il job può scriverci, ma il workflow lo elimina al termine del job. Per conservare i file dopo la fine del job, salvali nell'archiviazione file, che funziona come un tuo bucket di archiviazione. I file che carichi e i file che il tuo workflow salva rimangono nell'archiviazione file e le esecuzioni successive del workflow possono leggerli.

Ogni workflow ha una propria archiviazione file per ogni utente. Gli altri utenti che eseguono lo stesso workflow non vedono i tuoi file.

Carica un file di input

Carica un file dal tuo notebook locale con file_upload(), quindi elenca i file nell'archiviazione con files():

run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")

client.file_upload("hamiltonian.npz", my_workflow)
print(client.files(my_workflow))

['hamiltonian.npz']

Tipi di file supportati

Un workflow gestito controlla il contenuto di ogni file caricato, non la sua estensione, e rifiuta i file di altri tipi:

Accettati	Rifiutati
File JSON, di testo semplice, YAML e Python; file binari come NumPy (.npy, .npz), QPY e pickle; archivi tar, gzip e zip	File CSV, PNG, PDF, HDF5 e file vuoti

Nomi dei file

• Usa solo lettere, cifre, trattini bassi (_), trattini (-) e punti (.). Un workflow gestito rimuove dal nome gli altri caratteri, come gli spazi, quando scarichi o elimini un file. Ad esempio, puoi caricare my results.json, ma non puoi scaricarlo.

• L'archiviazione file non ha cartelle. file_upload() conserva solo il nome del file, quindi inputs/params.json viene memorizzato come params.json.

Leggere i file in un workflow

Passa il nome del file quando esegui il workflow:

run_workflow_remotely.ipynb

job = my_workflow.run(input_file="hamiltonian.npz")

Per abilitarlo, modifica il codice del tuo workflow in modo che legga il file da /function_user_data, dove ogni job vede l'archiviazione file:

source_files/my_workflow.py

from pathlib import Path

import numpy as np
from qiskit_serverless import get_arguments

FILE_STORAGE = Path("/function_user_data")

arguments = get_arguments()
hamiltonian = np.load(FILE_STORAGE / arguments["input_file"])

All'interno di un job, leggi e scrivi i file tramite questa cartella. I metodi dei file del client, come file_upload() e file_download(), sono per il tuo notebook locale.

Per leggere un archivio, estrailo sul disco locale:

source_files/my_workflow.py

import tarfile

with tarfile.open(FILE_STORAGE / "inputs.tar.gz") as tar:
    tar.extractall("/tmp/inputs", filter="data")

Scrivi i file temporanei in /tmp. L'archiviazione file è un'archiviazione a oggetti che il job vede come una cartella, quindi è più lenta del disco locale. Esegui le letture e le scritture pesanti sul disco locale e copia i file finiti nell'archiviazione file.

Salva i file di output

Il tuo notebook locale legge dal risultato i nomi dei file di output:

run_workflow_remotely.ipynb

result = job.result()
print(result["output_files"])  # ['energies.npy', 'summary.json']

Per abilitarlo, salva ogni file di output al livello principale di /function_user_data e restituisci i nomi dei file con save_result():

source_files/my_workflow.py

import json

from qiskit_serverless import save_result

np.save(FILE_STORAGE / "energies.npy", energies)
(FILE_STORAGE / "summary.json").write_text(json.dumps({"min_energy": float(energies.min())}))

save_result({"output_files": ["energies.npy", "summary.json"]})

I file che il job scrive fuori da /function_user_data, ad esempio in /tmp o nella cartella corrente, vengono eliminati al termine del job.

Scarica i file di output

Dopo la fine del job, scarica ogni file con file_download(). Imposta target_name per scegliere il nome del file locale e download_location per scegliere una cartella locale esistente:

run_workflow_remotely.ipynb

from pathlib import Path

Path("outputs").mkdir(exist_ok=True)

for name in result["output_files"]:
    client.file_download(name, my_workflow, target_name=name, download_location="outputs")

Senza target_name, file_download() aggiunge un prefisso casuale al nome del file locale, ad esempio downloaded_3f2a9c1b_energies.npy, e restituisce il nome che ha usato.

Scarica molti output come un unico archivio

Se un job salva molti file di output, o file in sottocartelle, raggruppali in un unico archivio nel workflow e scarica solo l'archivio:

run_workflow_remotely.ipynb

import tarfile

client.file_download(result["archive"], my_workflow, target_name=result["archive"], download_location="outputs")

with tarfile.open(Path("outputs") / result["archive"]) as tar:
    tar.extractall("outputs", filter="data")

Per abilitarlo, scrivi gli output in una cartella sul disco locale, aggiungi la cartella a un archivio in /tmp e copia l'archivio nell'archiviazione file:

source_files/my_workflow.py

import shutil
import tarfile

output_dir = Path("/tmp/outputs")
output_dir.mkdir()
np.save(output_dir / "energies.npy", energies)
# Save other outputs, including subfolders, to output_dir

archive_name = f"{run_name}_outputs.tar.gz"
with tarfile.open(Path("/tmp") / archive_name, "w:gz") as tar:
    tar.add(output_dir, arcname=run_name)
shutil.copy(Path("/tmp") / archive_name, FILE_STORAGE / archive_name)

save_result({"archive": archive_name})

Poiché l'archivio viene creato sul disco locale e copiato in un unico passaggio, l'archiviazione file non contiene mai un archivio incompleto. L'archivio mantiene la struttura delle cartelle e può contenere file di qualsiasi tipo. In questo esempio i file vengono estratti in outputs/<run_name>/.

Elimina i file

Un workflow gestito non elimina i file dall'archiviazione file, quindi i file restano finché non li elimini con file_delete():

run_workflow_remotely.ipynb

client.file_delete("hamiltonian.npz", my_workflow)

file_delete() segnala successo anche se il file non esiste, quindi controlla prima il nome con files(). Per eliminare tutti i file di un'esecuzione, filtra files() in base al prefisso del nome dell'esecuzione, come descritto nella sezione successiva:

run_workflow_remotely.ipynb

for name in client.files(my_workflow):
    if name.startswith("theta-0.1_"):
        client.file_delete(name, my_workflow)

Best practice: assegna a ogni file di un'esecuzione un nome univoco

Tutte le tue esecuzioni di un workflow condividono la stessa archiviazione file, e un file salvato da un job sostituisce qualsiasi file memorizzato con lo stesso nome. Se avvii più esecuzioni contemporaneamente e ognuna salva energies.npy, viene conservato solo l'ultimo.

Come best practice, passa un nome di esecuzione come argomento e inizia con esso il nome di ogni file salvato dall'esecuzione:

run_workflow_remotely.ipynb

jobs = {}
for theta in [0.1, 0.2, 0.3]:
    run_name = f"theta-{theta}"
    jobs[run_name] = my_workflow.run(run_name=run_name, theta=theta, input_file="hamiltonian.npz")

source_files/my_workflow.py

run_name = arguments["run_name"]
np.save(FILE_STORAGE / f"{run_name}_energies.npy", energies)

Tutte le esecuzioni possono leggere lo stesso file di input, hamiltonian.npz, perché lo leggono soltanto. Carica un input condiviso una sola volta, prima di avviare le esecuzioni.

Usa nomi di esecuzione che potrai riconoscere in seguito e che contengano solo lettere, cifre, caratteri di sottolineatura, trattini e punti. Con un prefisso per ogni esecuzione puoi trovare, scaricare ed eliminare insieme i file di un'esecuzione.

Best practice: usa checkpoint per i job lunghi

Un job che viene annullato, fallisce o raggiunge il limite di 24 ore perde tutto ciò che non si trova nell'archiviazione file. Per le esecuzioni lunghe, salva un checkpoint nell'archiviazione file a intervalli regolari e riprendi da esso quando esegui di nuovo il workflow con lo stesso nome di esecuzione:

source_files/my_workflow.py

import json
import shutil

checkpoint = FILE_STORAGE / f"{run_name}_checkpoint.json"

state = {"step": 0, "params": initial_params}
if checkpoint.exists():
    state = json.loads(checkpoint.read_text())

for step in range(state["step"], num_steps):
    state["params"] = optimize_step(state["params"])
    state["step"] = step + 1

    if state["step"] % 10 == 0:
        local_copy = Path("/tmp") / checkpoint.name
        local_copy.write_text(json.dumps(state))
        shutil.copy(local_copy, checkpoint)

Per continuare un'esecuzione interrotta, chiama di nuovo run() con gli stessi argomenti.

Tieni presente quanto segue:

• Scrivi prima il checkpoint sul disco locale, poi copialo. Se il job si interrompe durante una scrittura, il checkpoint memorizzato è incompleto e l'esecuzione successiva non può leggerlo.

• Scegli un intervallo adatto al costo di un passo. Salva abbastanza spesso da non perdere troppo lavoro, ma non così spesso che la scrittura dei checkpoint rallenti il job.

• Elimina i checkpoint dopo il successo dell'esecuzione. Un'esecuzione successiva con lo stesso nome ripartirà quindi dall'inizio.

Per un job annullato, puoi anche salvare i risultati parziali in un gestore SIGTERM.

Passi successivi

Raccomandazioni

• Esegui uno sweep di parametri per avviare un job per ogni input.
• Scrivi log e gestisci gli errori per salvare i risultati parziali quando un job viene annullato.