Dateien über mehrere Läufe hinweg verwalten

Ein verwalteter Workflow-Job funktioniert wie ein Batch-Job auf einem HPC-Rechenknoten. Der lokale Datenträger des Jobs ist Scratch-Speicher. Der Job kann darauf schreiben, aber der Workflow löscht ihn, wenn der Job endet. Um Dateien über das Ende des Jobs hinaus zu behalten, speichere sie im Dateispeicher, der wie dein eigener Storage-Bucket funktioniert. Dateien, die du hochlädst, und Dateien, die dein Workflow speichert, bleiben im Dateispeicher, und spätere Läufe des Workflows können sie lesen.

Jeder Workflow hat für jeden Benutzer einen eigenen Dateispeicher. Andere Benutzer, die denselben Workflow ausführen, sehen deine Dateien nicht.

Eine Eingabedatei hochladen

Lade eine Datei aus deinem lokalen Notebook mit file_upload() hoch und liste dann die Dateien im Speicher mit files() auf:

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']

Unterstützte Dateitypen

Ein verwalteter Workflow prüft den Inhalt jeder hochgeladenen Datei, nicht ihre Erweiterung, und lehnt Dateien anderer Typen ab:

Akzeptiert	Abgelehnt
JSON-, Klartext-, YAML- und Python-Dateien; Binärdateien wie NumPy- (.npy, .npz), QPY- und Pickle-Dateien; tar-, gzip- und zip-Archive	CSV-, PNG-, PDF-, HDF5- und leere Dateien

Dateinamen

• Verwende nur Buchstaben, Ziffern, Unterstriche (_), Bindestriche (-) und Punkte (.). Ein verwalteter Workflow entfernt andere Zeichen, etwa Leerzeichen, aus dem Namen, wenn du eine Datei herunterlädst oder löschst. Du kannst zum Beispiel my results.json hochladen, aber nicht herunterladen.

• Der Dateispeicher hat keine Ordner. file_upload() behält nur den Dateinamen, sodass inputs/params.json als params.json gespeichert wird.

Dateien in einem Workflow lesen

Übergib den Namen der Datei, wenn du den Workflow ausführst:

run_workflow_remotely.ipynb

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

Um dies zu ermöglichen, bearbeite deinen Workflow-Code so, dass er die Datei aus /function_user_data liest, wo jeder Job den Dateispeicher sieht:

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

Lies und schreibe Dateien innerhalb eines Jobs über diesen Ordner. Die Client-Dateimethoden wie file_upload() und file_download() sind für dein lokales Notebook gedacht.

Um ein Archiv zu lesen, extrahiere es auf den lokalen Datenträger:

source_files/my_workflow.py

import tarfile

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

Schreibe temporäre Dateien nach /tmp. Der Dateispeicher ist ein Objektspeicher, den der Job als Ordner sieht, daher ist er langsamer als der lokale Datenträger. Führe umfangreiche Lese- und Schreibvorgänge auf dem lokalen Datenträger aus und kopiere fertige Dateien in den Dateispeicher.

Ausgabedateien speichern

Dein lokales Notebook liest die Namen der Ausgabedateien aus dem Ergebnis:

run_workflow_remotely.ipynb

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

Um dies zu ermöglichen, speichere jede Ausgabedatei auf der obersten Ebene von /function_user_data und gib die Dateinamen mit save_result() zurück:

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

Dateien, die der Job außerhalb von /function_user_data schreibt, etwa in /tmp oder im aktuellen Ordner, werden gelöscht, wenn der Job endet.

Ausgabedateien herunterladen

Lade nach dem Ende des Jobs jede Datei mit file_download() herunter. Setze target_name, um den lokalen Dateinamen zu wählen, und download_location, um einen vorhandenen lokalen Ordner zu wählen:

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

Ohne target_name stellt file_download() dem lokalen Dateinamen ein zufälliges Präfix voran, etwa downloaded_3f2a9c1b_energies.npy, und gibt den verwendeten Namen zurück.

Viele Ausgaben als ein Archiv herunterladen

Wenn ein Job viele Ausgabedateien oder Dateien in Unterordnern speichert, bündle sie im Workflow in einem Archiv und lade nur das Archiv herunter:

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

Dafür schreibst du die Ausgaben in einen Ordner auf dem lokalen Datenträger, fügst den Ordner einem Archiv in /tmp hinzu und kopierst das Archiv in den Dateispeicher:

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

Da das Archiv auf dem lokalen Datenträger erstellt und in einem Schritt kopiert wird, enthält der Dateispeicher nie ein unvollständiges Archiv. Das Archiv behält die Ordnerstruktur bei und kann Dateien beliebigen Typs enthalten. In diesem Beispiel werden die Dateien nach outputs/<run_name>/ extrahiert.

Dateien löschen

Ein verwalteter Workflow löscht keine Dateien aus dem Dateispeicher. Die Dateien bleiben daher erhalten, bis du sie mit file_delete() löschst:

run_workflow_remotely.ipynb

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

file_delete() meldet auch dann Erfolg, wenn die Datei nicht existiert. Prüfe den Namen deshalb zuerst mit files(). Um alle Dateien eines Laufs zu löschen, filterst du files() nach dem Namenspräfix des Laufs, wie im nächsten Abschnitt beschrieben:

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: Gib den Dateien jedes Laufs einen eindeutigen Namen

Alle deine Läufe eines Workflows teilen sich denselben Dateispeicher, und eine Datei, die ein Job speichert, ersetzt jede gespeicherte Datei mit demselben Namen. Wenn du mehrere Läufe gleichzeitig startest und jeder Lauf energies.npy speichert, bleibt nur die zuletzt gespeicherte Datei erhalten.

Als Best Practice übergibst du einen Lauf-Namen als Argument und beginnst den Namen jeder Datei, die der Lauf speichert, mit diesem Namen:

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)

Alle Läufe können dieselbe Eingabedatei hamiltonian.npz lesen, da sie sie nur lesen. Lade eine gemeinsame Eingabe einmal hoch, bevor du die Läufe startest.

Verwende Lauf-Namen, die du später wiedererkennst und die nur Buchstaben, Ziffern, Unterstriche, Bindestriche und Punkte enthalten. Mit einem Präfix pro Lauf kannst du die Dateien eines Laufs gemeinsam finden, herunterladen und löschen.

Best Practice: Setze Checkpoints für lange Jobs

Ein Job, der abgebrochen wird, fehlschlägt oder das 24-Stunden-Limit erreicht, verliert alles, was nicht im Dateispeicher liegt. Speichere bei langen Läufen in regelmäßigen Abständen einen Checkpoint im Dateispeicher und setze daran fort, wenn du den Workflow mit demselben Lauf-Namen erneut ausführst:

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)

Um einen gestoppten Lauf fortzusetzen, rufe run() erneut mit denselben Argumenten auf.

Beachte die folgenden Punkte:

• Schreibe den Checkpoint zuerst auf den lokalen Datenträger und kopiere ihn dann. Wenn der Job während eines Schreibvorgangs stoppt, ist der gespeicherte Checkpoint unvollständig, und der nächste Lauf kann ihn nicht lesen.

• Wähle ein Intervall, das zu den Kosten eines Schritts passt. Speichere oft genug, damit nicht viel Arbeit verloren geht, aber nicht so oft, dass das Schreiben der Checkpoints den Job verlangsamt.

• Lösche Checkpoints, nachdem der Lauf erfolgreich war. Ein späterer Lauf mit demselben Namen beginnt dann von vorn.

Bei einem Job, der abgebrochen wird, kannst du Teilergebnisse auch in einem SIGTERM handler speichern.

Nächste Schritte

Empfehlungen

• Führe einen Parameter-Sweep aus, um für jede Eingabe einen Job zu starten.
• Schreibe Logs und behandle Fehler, um Teilergebnisse zu speichern, wenn ein Job abgebrochen wird.