Gérer les fichiers d'une exécution à l'autre

Un job de workflow géré fonctionne comme un job batch sur un nœud de calcul HPC. Le disque local du job est un espace temporaire. Le job peut y écrire, mais le workflow le supprime à la fin du job. Pour conserver des fichiers après la fin du job, enregistre-les dans le stockage de fichiers, qui fonctionne comme ton propre bucket de stockage. Les fichiers que tu téléverses, et ceux que ton workflow enregistre, restent dans le stockage de fichiers, et les exécutions ultérieures du workflow peuvent les lire.

Chaque workflow dispose de son propre stockage de fichiers pour chaque utilisateur. Les autres utilisateurs qui exécutent le même workflow ne voient pas tes fichiers.

Téléverser un fichier d'entrée

Téléverse un fichier depuis ton notebook local avec file_upload(), puis liste les fichiers du stockage avec 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']

Types de fichiers pris en charge

Un workflow géré vérifie le contenu de chaque fichier téléversé, et non son extension, et rejette les fichiers d'autres types :

Acceptés	Rejetés
Fichiers JSON, texte brut, YAML et Python ; fichiers binaires tels que NumPy (.npy, .npz), QPY et pickle ; archives tar, gzip et zip	Fichiers CSV, PNG, PDF, HDF5 et fichiers vides

Noms de fichiers

• Utilise uniquement des lettres, des chiffres, des tirets bas (_), des tirets (-) et des points (.). Un workflow géré supprime les autres caractères, comme les espaces, du nom quand tu télécharges ou supprimes un fichier. Par exemple, tu peux téléverser my results.json, mais tu ne peux pas le télécharger.

• Le stockage de fichiers n'a pas de dossiers. file_upload() ne conserve que le nom du fichier, donc inputs/params.json est stocké sous params.json.

Lire des fichiers dans un workflow

Passe le nom du fichier quand tu exécutes le workflow :

run_workflow_remotely.ipynb

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

Pour l'activer, modifie le code de ton workflow afin qu'il lise le fichier depuis /function_user_data, l'endroit où chaque job voit le stockage de fichiers :

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"])

À l'intérieur d'un job, lis et écris les fichiers via ce dossier. Les méthodes de fichiers du client, comme file_upload() et file_download(), sont destinées à ton notebook local.

Pour lire une archive, extrais-la sur le disque local :

source_files/my_workflow.py

import tarfile

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

Écris les fichiers temporaires dans /tmp. Le stockage de fichiers est un stockage objet que le job voit comme un dossier ; il est donc plus lent que le disque local. Effectue les lectures et écritures volumineuses sur le disque local, puis copie les fichiers terminés vers le stockage de fichiers.

Enregistrer les fichiers de sortie

Ton notebook local lit les noms des fichiers de sortie à partir du résultat :

run_workflow_remotely.ipynb

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

Pour cela, enregistre chaque fichier de sortie au niveau supérieur de /function_user_data, et renvoie les noms de fichiers avec 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"]})

Les fichiers que le job écrit en dehors de /function_user_data, par exemple dans /tmp ou dans le dossier courant, sont supprimés à la fin du job.

Télécharger les fichiers de sortie

Une fois le job terminé, télécharge chaque fichier avec file_download(). Définis target_name pour choisir le nom du fichier local, et download_location pour choisir un dossier local existant :

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

Sans target_name, file_download() ajoute un préfixe aléatoire au nom du fichier local, par exemple downloaded_3f2a9c1b_energies.npy, et renvoie le nom qu'il a utilisé.

Télécharger plusieurs sorties sous forme d'une seule archive

Si un job enregistre de nombreux fichiers de sortie, ou des fichiers dans des sous-dossiers, regroupe-les dans une seule archive dans le workflow, et télécharge uniquement l'archive :

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

Pour cela, écris les sorties dans un dossier du disque local, ajoute ce dossier à une archive dans /tmp, puis copie l'archive vers le stockage de fichiers :

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

Comme l'archive est construite sur le disque local et copiée en une seule étape, le stockage de fichiers ne contient jamais d'archive incomplète. L'archive conserve la structure des dossiers et peut contenir des fichiers de tout type. Dans cet exemple, les fichiers sont extraits dans outputs/<run_name>/.

Supprimer des fichiers

Un workflow géré ne supprime pas les fichiers du stockage de fichiers ; ils restent donc jusqu'à ce que tu les supprimes avec file_delete() :

run_workflow_remotely.ipynb

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

file_delete() signale un succès même si le fichier n'existe pas ; vérifie donc d'abord le nom avec files(). Pour supprimer tous les fichiers d'une exécution, filtre files() par le préfixe du nom de l'exécution, comme décrit dans la section suivante :

run_workflow_remotely.ipynb

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

Bonne pratique : donne un nom unique aux fichiers de chaque exécution

Toutes tes exécutions d'un workflow partagent le même stockage de fichiers, et un fichier qu'un job enregistre remplace tout fichier stocké portant le même nom. Si tu lances plusieurs exécutions en même temps et que chacune enregistre energies.npy, seul le dernier est conservé.

Comme bonne pratique, passe un nom d'exécution en argument, et fais commencer par ce nom chaque fichier que l'exécution enregistre :

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)

Toutes les exécutions peuvent lire le même fichier d'entrée, hamiltonian.npz, car elles ne font que le lire. Téléverse une entrée partagée une seule fois, avant de lancer les exécutions.

Utilise des noms d'exécution que tu pourras reconnaître plus tard, et qui ne contiennent que des lettres, des chiffres, des tirets bas, des traits d'union et des points. Avec un préfixe par exécution, tu peux trouver, télécharger et supprimer ensemble les fichiers d'une exécution.

Bonne pratique : crée des points de contrôle pour les jobs longs

Un job qui est annulé, échoue ou atteint la limite de 24 heures perd tout ce qui n'est pas dans le stockage de fichiers. Pour les exécutions longues, enregistre un point de contrôle dans le stockage de fichiers à intervalles réguliers, et reprends à partir de celui-ci lorsque tu relances le workflow avec le même nom d'exécution :

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)

Pour poursuivre une exécution qui s'est arrêtée, appelle de nouveau run() avec les mêmes arguments.

Garde les points suivants à l'esprit :

• Écris d'abord le point de contrôle sur le disque local, puis copie-le. Si le job s'arrête pendant une écriture, le point de contrôle stocké est incomplet et l'exécution suivante ne peut pas le lire.

• Choisis un intervalle adapté au coût d'une étape. Enregistre assez souvent pour ne pas perdre beaucoup de travail, mais pas trop souvent pour que l'écriture des points de contrôle ne ralentisse pas le job.

• Supprime les points de contrôle une fois l'exécution réussie. Une exécution ultérieure portant le même nom repart alors du début.

Pour un job annulé, tu peux aussi enregistrer des résultats partiels dans un gestionnaire SIGTERM.

Étapes suivantes

Recommandations

• Exécuter un balayage de paramètres pour lancer un job par entrée.
• Écrire des journaux et gérer les erreurs pour enregistrer des résultats partiels lorsqu'un job est annulé.