Gestionează fișierele între rulări

Un job de flux de lucru gestionat funcționează ca un job batch pe un nod de calcul HPC. Discul local al jobului este spațiu temporar. Jobul poate scrie pe el, dar fluxul de lucru îl șterge când jobul se termină. Pentru a păstra fișierele după terminarea jobului, salvează-le în stocarea de fișiere, care funcționează ca propriul tău bucket de stocare. Fișierele pe care le încarci și fișierele pe care le salvează fluxul tău de lucru rămân în stocarea de fișiere, iar rulările ulterioare ale fluxului de lucru le pot citi.

Fiecare flux de lucru are propria stocare de fișiere pentru fiecare utilizator. Alți utilizatori care rulează același flux de lucru nu îți văd fișierele.

Încarcă un fișier de intrare

Încarcă un fișier din notebook-ul tău local cu file_upload(), apoi listează fișierele din stocare cu 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']

Tipuri de fișiere acceptate

Un flux de lucru gestionat verifică conținutul fiecărui fișier încărcat, nu extensia lui, și respinge fișierele de alte tipuri:

Acceptate	Respinse
Fișiere JSON, text simplu, YAML și Python; fișiere binare precum NumPy (.npy, .npz), QPY și pickle; arhive tar, gzip și zip	Fișiere CSV, PNG, PDF, HDF5 și fișiere goale

Nume de fișiere

• Folosește doar litere, cifre, liniuțe de subliniere (_), cratime (-) și puncte (.). Un flux de lucru gestionat elimină din nume alte caractere, precum spațiile, când descarci sau ștergi un fișier. De exemplu, poți încărca my results.json, dar nu îl poți descărca.

• Stocarea de fișiere nu are foldere. file_upload() păstrează doar numele fișierului, deci inputs/params.json este stocat ca params.json.

Citește fișiere într-un flux de lucru

Transmite numele fișierului când rulezi fluxul de lucru:

run_workflow_remotely.ipynb

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

Pentru a activa acest lucru, editează codul fluxului tău de lucru astfel încât să citească fișierul din /function_user_data, unde fiecare job vede stocarea de fișiere:

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

Într-un job, citește și scrie fișiere prin acest folder. Metodele client pentru fișiere, precum file_upload() și file_download(), sunt pentru notebook-ul tău local.

Pentru a citi o arhivă, extrage-o pe discul local:

source_files/my_workflow.py

import tarfile

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

Scrie fișierele temporare în /tmp. Stocarea de fișiere este o stocare de obiecte pe care jobul o vede ca pe un folder, deci este mai lentă decât discul local. Fă citirile și scrierile intensive pe discul local și copiază fișierele finalizate în stocarea de fișiere.

Salvează fișierele de ieșire

Notebook-ul tău local citește numele fișierelor de ieșire din rezultat:

run_workflow_remotely.ipynb

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

Pentru a activa acest lucru, salvează fiecare fișier de ieșire la nivelul superior al /function_user_data și returnează numele fișierelor cu 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"]})

Fișierele pe care jobul le scrie în afara /function_user_data, de exemplu în /tmp sau în folderul curent, sunt șterse când jobul se termină.

Descarcă fișierele de ieșire

După ce jobul se termină, descarcă fiecare fișier cu file_download(). Setează target_name pentru a alege numele fișierului local și download_location pentru a alege un folder local existent:

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

Fără target_name, file_download() adaugă un prefix aleatoriu la numele fișierului local, cum ar fi downloaded_3f2a9c1b_energies.npy, și returnează numele pe care l-a folosit.

Descarcă mai multe ieșiri ca o singură arhivă

Dacă un job salvează multe fișiere de ieșire sau fișiere în subfoldere, grupează-le într-o singură arhivă în workflow și descarcă doar arhiva:

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

Pentru a activa acest lucru, scrie ieșirile într-un folder de pe discul local, adaugă folderul într-o arhivă în /tmp și copiază arhiva în stocarea de fișiere:

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

Deoarece arhiva este construită pe discul local și copiată într-un singur pas, stocarea de fișiere nu conține niciodată o arhivă incompletă. Arhiva păstrează structura folderelor și poate conține fișiere de orice tip. În acest exemplu, fișierele sunt extrase în outputs/<run_name>/.

Șterge fișiere

Un workflow gestionat nu șterge fișiere din stocarea de fișiere, deci fișierele rămân până când le ștergi cu file_delete():

run_workflow_remotely.ipynb

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

file_delete() raportează succes chiar dacă fișierul nu există, așa că verifică mai întâi numele cu files(). Pentru a șterge toate fișierele unei rulări, filtrează files() după prefixul de nume al rulării, așa cum este descris în secțiunea următoare:

run_workflow_remotely.ipynb

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

Bună practică: dă fișierelor fiecărei rulări un nume unic

Toate rulările unui workflow folosesc aceeași stocare de fișiere, iar un fișier salvat de un job înlocuiește orice fișier stocat cu același nume. Dacă pornești mai multe rulări în același timp și fiecare rulare salvează energies.npy, doar ultimul este păstrat.

Ca bună practică, transmite un nume de rulare ca argument și începe cu el numele fiecărui fișier pe care îl salvează rularea:

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)

Toate rulările pot citi același fișier de intrare, hamiltonian.npz, deoarece doar îl citesc. Încarcă o intrare comună o singură dată, înainte de a porni rulările.

Folosește nume de rulare pe care le poți recunoaște mai târziu și care conțin doar litere, cifre, liniuțe de subliniere, cratime și puncte. Cu un prefix pentru fiecare rulare, poți găsi, descărca și șterge împreună fișierele unei rulări.

Bună practică: creează puncte de control pentru joburile lungi

Un job care este anulat, eșuează sau atinge limita de 24 de ore pierde tot ce nu se află în stocarea de fișiere. Pentru rulări lungi, salvează un punct de control în stocarea de fișiere la intervale regulate și reia de la el când rulezi din nou workflow-ul cu același nume de rulare:

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)

Pentru a continua o rulare care s-a oprit, apelează din nou run() cu aceleași argumente.

Ține cont de următoarele puncte:

• Scrie mai întâi punctul de control pe discul local, apoi copiază-l. Dacă jobul se oprește în timpul unei scrieri, punctul de control stocat este incomplet, iar rularea următoare nu îl poate citi.

• Alege un interval care corespunde costului unui pas. Salvează suficient de des încât să nu pierzi prea multă muncă, dar nu atât de des încât scrierea punctelor de control să încetinească jobul.

• Șterge punctele de control după ce rularea reușește. O rulare ulterioară cu același nume va porni atunci de la început.

Pentru un job care este anulat, poți salva și rezultate parțiale într-un handler SIGTERM.

Pașii următori

Recomandări

• Rulează o parcurgere de parametri pentru a porni un job pentru fiecare intrare.
• Scrie loguri și gestionează erorile pentru a salva rezultate parțiale când un job este anulat.