Gestionar archivos entre ejecuciones

Un trabajo de flujo de trabajo gestionado funciona como un trabajo por lotes en un nodo de cómputo HPC. El disco local del trabajo es espacio temporal. El trabajo puede escribir en él, pero el flujo de trabajo lo elimina cuando el trabajo termina. Para conservar archivos después de que el trabajo termine, guárdalos en el almacenamiento de archivos, que funciona como tu propio bucket de almacenamiento. Los archivos que subes y los archivos que guarda tu flujo de trabajo permanecen en el almacenamiento de archivos, y las ejecuciones posteriores del flujo de trabajo pueden leerlos.

Cada flujo de trabajo tiene su propio almacenamiento de archivos para cada usuario. Otros usuarios que ejecuten el mismo flujo de trabajo no ven tus archivos.

Sube un archivo de entrada

Sube un archivo desde tu notebook local con file_upload() y luego enumera los archivos del almacenamiento 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']

Tipos de archivo admitidos

Un flujo de trabajo gestionado comprueba el contenido de cada archivo subido, no su extensión, y rechaza los archivos de otros tipos:

Aceptados	Rechazados
Archivos JSON, de texto sin formato, YAML y Python; archivos binarios como los de NumPy (.npy, .npz), QPY y pickle; archivos tar, gzip y zip	Archivos CSV, PNG, PDF, HDF5 y archivos vacíos

Nombres de archivo

• Usa solo letras, dígitos, guiones bajos (_), guiones (-) y puntos (.). Un flujo de trabajo gestionado elimina del nombre otros caracteres, como los espacios, cuando descargas o eliminas un archivo. Por ejemplo, puedes subir my results.json, pero no puedes descargarlo.

• El almacenamiento de archivos no tiene carpetas. file_upload() conserva solo el nombre del archivo, así que inputs/params.json se almacena como params.json.

Lee archivos en un flujo de trabajo

Pasa el nombre del archivo cuando ejecutes el flujo de trabajo:

run_workflow_remotely.ipynb

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

Para habilitarlo, edita el código de tu flujo de trabajo para que lea el archivo desde /function_user_data, que es donde cada trabajo ve el almacenamiento de archivos:

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

Dentro de un trabajo, lee y escribe archivos a través de esta carpeta. Los métodos de archivo del cliente, como file_upload() y file_download(), son para tu notebook local.

Para leer un archivo comprimido, extráelo en el disco local:

source_files/my_workflow.py

import tarfile

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

Escribe los archivos temporales en /tmp. El almacenamiento de archivos es un almacenamiento de objetos que el trabajo ve como una carpeta, por lo que es más lento que el disco local. Realiza las lecturas y escrituras intensivas en el disco local y copia los archivos terminados en el almacenamiento de archivos.

Guarda los archivos de salida

Tu notebook local lee los nombres de los archivos de salida a partir del resultado:

run_workflow_remotely.ipynb

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

Para habilitar esto, guarda cada archivo de salida en el nivel superior de /function_user_data y devuelve los nombres de los archivos 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"]})

Los archivos que el trabajo escribe fuera de /function_user_data, por ejemplo en /tmp o en la carpeta actual, se eliminan cuando el trabajo termina.

Descargar archivos de salida

Después de que el trabajo termine, descarga cada archivo con file_download(). Establece target_name para elegir el nombre del archivo local y download_location para elegir una carpeta local existente:

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

Sin target_name, file_download() añade un prefijo aleatorio al nombre del archivo local, como downloaded_3f2a9c1b_energies.npy, y devuelve el nombre que usó.

Descargar muchas salidas como un solo archivo comprimido

Si un trabajo guarda muchos archivos de salida, o archivos en subcarpetas, agrúpalos en un solo archivo comprimido dentro del workflow y descarga solo ese archivo comprimido:

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

Para habilitar esto, escribe las salidas en una carpeta del disco local, añade la carpeta a un archivo comprimido en /tmp y copia el archivo comprimido al almacenamiento de archivos:

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

Como el archivo comprimido se construye en el disco local y se copia en un solo paso, el almacenamiento de archivos nunca contiene un archivo comprimido incompleto. El archivo comprimido conserva la estructura de carpetas y puede contener archivos de cualquier tipo. En este ejemplo, los archivos se extraen en outputs/<run_name>/.

Eliminar archivos

Un workflow gestionado no elimina archivos del almacenamiento de archivos, así que los archivos permanecen hasta que los elimines con file_delete():

run_workflow_remotely.ipynb

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

file_delete() informa de éxito aunque el archivo no exista, así que comprueba antes el nombre con files(). Para eliminar todos los archivos de una ejecución, filtra files() por el prefijo del nombre de la ejecución, como se describe en la siguiente sección:

run_workflow_remotely.ipynb

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

Buena práctica: da a los archivos de cada ejecución un nombre único

Todas tus ejecuciones de un workflow comparten el mismo almacenamiento de archivos, y un archivo que guarda un trabajo reemplaza cualquier archivo almacenado con el mismo nombre. Si inicias varias ejecuciones al mismo tiempo y cada una guarda energies.npy, solo se conserva la última.

Como buena práctica, pasa un nombre de ejecución como argumento y empieza con él el nombre de cada archivo que guarde la ejecución:

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)

Todas las ejecuciones pueden leer el mismo archivo de entrada, hamiltonian.npz, porque solo lo leen. Sube una entrada compartida una sola vez, antes de iniciar las ejecuciones.

Usa nombres de ejecución que puedas reconocer más adelante y que contengan solo letras, dígitos, guiones bajos, guiones y puntos. Con un prefijo para cada ejecución, puedes encontrar, descargar y eliminar juntos los archivos de una ejecución.

Buena práctica: crea puntos de control en los trabajos largos

Un trabajo que se cancela, falla o alcanza el límite de 24 horas pierde todo lo que no está en el almacenamiento de archivos. En ejecuciones largas, guarda un punto de control en el almacenamiento de archivos a intervalos regulares y reanuda desde él cuando vuelvas a ejecutar el workflow con el mismo nombre de ejecución:

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)

Para continuar una ejecución que se detuvo, llama de nuevo a run() con los mismos argumentos.

Ten en cuenta los siguientes puntos:

• Escribe primero el punto de control en el disco local y luego cópialo. Si el trabajo se detiene durante una escritura, el punto de control almacenado queda incompleto y la siguiente ejecución no puede leerlo.

• Elige un intervalo acorde con el costo de cada paso. Guarda con la frecuencia suficiente para no perder mucho trabajo, pero no tan a menudo que escribir los puntos de control ralentice el trabajo.

• Elimina los puntos de control después de que la ejecución termine con éxito. Así, una ejecución posterior con el mismo nombre empieza desde el principio.

En un trabajo que se cancela, también puedes guardar resultados parciales en un controlador de SIGTERM.

Próximos pasos

Recomendaciones

• Ejecuta un barrido de parámetros para iniciar un trabajo por cada entrada.
• Escribe registros y gestiona errores para guardar resultados parciales cuando se cancela un trabajo.