Správa souborů napříč běhy

Úloha spravovaného workflow funguje jako dávková úloha na výpočetním uzlu HPC. Lokální disk úlohy je odkládací prostor. Úloha na něj může zapisovat, ale workflow jej smaže, když úloha skončí. Chceš-li soubory zachovat i po skončení úlohy, ulož je do úložiště souborů, které funguje jako tvůj vlastní storage bucket. Soubory, které nahraješ, a soubory, které tvůj workflow uloží, zůstávají v úložišti souborů a pozdější běhy workflow je mohou číst.

Každý workflow má pro každého uživatele vlastní úložiště souborů. Ostatní uživatelé, kteří spouštějí stejný workflow, tvoje soubory nevidí.

Nahrání vstupního souboru

Nahraj soubor ze svého lokálního notebooku pomocí file_upload() a poté vypiš soubory v úložišti pomocí 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']

Podporované typy souborů

Spravovaný workflow kontroluje obsah každého nahraného souboru, nikoli jeho příponu, a soubory jiných typů odmítá:

Přijímané	Odmítané
Soubory JSON, prostý text, YAML a Python; binární soubory jako NumPy (.npy, .npz), QPY a pickle; archivy tar, gzip a zip	Soubory CSV, PNG, PDF, HDF5 a prázdné soubory

Názvy souborů

• Používej pouze písmena, číslice, podtržítka (_), pomlčky (-) a tečky (.). Spravovaný workflow z názvu odstraní jiné znaky, například mezery, když soubor stahuješ nebo mažeš. Například můžeš nahrát my results.json, ale nemůžeš ho stáhnout.

• Úložiště souborů nemá žádné složky. file_upload() ponechá pouze název souboru, takže inputs/params.json se uloží jako params.json.

Čtení souborů ve workflow

Při spuštění workflow předej název souboru:

run_workflow_remotely.ipynb

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

Abys to umožnil, uprav kód svého workflow tak, aby četl soubor z /function_user_data, což je místo, kde každá úloha vidí úložiště souborů:

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

Uvnitř úlohy čti a zapisuj soubory přes tuto složku. Klientské metody pro soubory, například file_upload() a file_download(), jsou určené pro tvůj lokální notebook.

Chceš-li přečíst archiv, rozbal ho na lokální disk:

source_files/my_workflow.py

import tarfile

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

Dočasné soubory zapisuj do /tmp. Úložiště souborů je objektové úložiště, které úloha vidí jako složku, takže je pomalejší než lokální disk. Náročné čtení a zápis provádějte na lokálním disku a hotové soubory zkopíruj do úložiště souborů.

Ulož výstupní soubory

Tvůj lokální notebook čte názvy výstupních souborů z výsledku:

run_workflow_remotely.ipynb

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

Abys to umožnil, ulož každý výstupní soubor na nejvyšší úroveň /function_user_data a vrať názvy souborů pomocí 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"]})

Soubory, které úloha zapíše mimo /function_user_data, například do /tmp nebo do aktuální složky, se po skončení úlohy smažou.

Stáhni výstupní soubory

Po skončení úlohy stáhni každý soubor pomocí file_download(). Pomocí target_name zvolíš název lokálního souboru a pomocí download_location zvolíš existující lokální složku:

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

Bez target_name přidá file_download() k názvu lokálního souboru náhodnou předponu, například downloaded_3f2a9c1b_energies.npy, a vrátí název, který použila.

Stáhni mnoho výstupů jako jeden archiv

Pokud úloha ukládá mnoho výstupních souborů nebo soubory v podsložkách, sbal je ve workflow do jednoho archivu a stáhni pouze archiv:

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

Abys to umožnil, zapiš výstupy do složky na lokálním disku, přidej složku do archivu v /tmp a zkopíruj archiv do úložiště souborů:

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

Protože se archiv vytvoří na lokálním disku a zkopíruje se jedním krokem, úložiště souborů nikdy neobsahuje neúplný archiv. Archiv zachovává strukturu složek a může obsahovat soubory libovolného typu. V tomto příkladu se soubory rozbalí do outputs/<run_name>/.

Smaž soubory

Spravovaný workflow nemaže soubory z úložiště souborů, takže soubory zůstanou, dokud je nesmažeš pomocí file_delete():

run_workflow_remotely.ipynb

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

file_delete() hlásí úspěch i tehdy, když soubor neexistuje, proto nejprve zkontroluj název pomocí files(). Chceš-li smazat všechny soubory jednoho běhu, vyfiltruj files() podle předpony názvu běhu, jak je popsáno v další části:

run_workflow_remotely.ipynb

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

Osvědčený postup: dej souborům každého běhu jedinečný název

Všechny tvé běhy jednoho workflow sdílejí stejné úložiště souborů a soubor, který úloha uloží, nahradí každý uložený soubor se stejným názvem. Pokud spustíš několik běhů současně a každý běh uloží energies.npy, zůstane zachován pouze poslední.

Jako osvědčený postup předej název běhu jako argument a každý soubor, který běh ukládá, začni tímto názvem:

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)

Všechny běhy mohou číst stejný vstupní soubor hamiltonian.npz, protože ho pouze čtou. Sdílený vstup nahraj jednou, před spuštěním běhů.

Používej názvy běhů, které později poznáš a které obsahují pouze písmena, číslice, podtržítka, pomlčky a tečky. Díky jedné předponě pro každý běh můžeš soubory běhu společně najít, stáhnout a smazat.

Osvědčený postup: ukládej kontrolní body u dlouhých úloh

Úloha, která je zrušena, selže nebo dosáhne 24hodinového limitu, ztratí vše, co není v úložišti souborů. U dlouhých běhů ukládej kontrolní bod do úložiště souborů v pravidelných intervalech a při dalším spuštění workflow se stejným názvem běhu z něj pokračuj:

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)

Chceš-li pokračovat v běhu, který se zastavil, zavolej znovu run() se stejnými argumenty.

Mysli na následující body:

• Kontrolní bod nejprve zapiš na lokální disk a poté ho zkopíruj. Pokud se úloha zastaví během zápisu, uložený kontrolní bod je neúplný a další běh ho nemůže přečíst.

• Zvol interval, který odpovídá nákladům jednoho kroku. Ukládej dost často, abys neztratil mnoho práce, ale ne tak často, aby zápis kontrolních bodů úlohu zpomaloval.

• Po úspěšném dokončení běhu smaž kontrolní body. Pozdější běh se stejným názvem pak začne od začátku.

U úlohy, která je zrušena, můžeš také uložit částečné výsledky v obslužné funkci SIGTERM.

Další kroky

Doporučení

• Spusť prohledávání parametrů a spusť jednu úlohu pro každý vstup.
• Zapisuj logy a ošetřuj chyby a ukládej částečné výsledky, když je úloha zrušena.