Zarządzanie plikami między uruchomieniami

Zadanie zarządzanego przepływu pracy działa jak zadanie wsadowe na węźle obliczeniowym HPC. Lokalny dysk zadania to przestrzeń robocza. Zadanie może na nim zapisywać, ale przepływ pracy usuwa go po zakończeniu zadania. Aby zachować pliki po zakończeniu zadania, zapisz je w magazynie plików, który działa jak twój własny kosz na dane. Pliki, które przesyłasz, i pliki zapisywane przez twój przepływ pracy pozostają w magazynie plików, a późniejsze uruchomienia przepływu pracy mogą je odczytać.

Każdy przepływ pracy ma własny magazyn plików dla każdego użytkownika. Inni użytkownicy uruchamiający ten sam przepływ pracy nie widzą twoich plików.

Prześlij plik wejściowy

Prześlij plik z lokalnego notebooka za pomocą file_upload(), a następnie wyświetl pliki w magazynie za 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']

Obsługiwane typy plików

Zarządzany przepływ pracy sprawdza zawartość każdego przesłanego pliku, a nie jego rozszerzenie, i odrzuca pliki innych typów:

Akceptowane	Odrzucane
Pliki JSON, zwykły tekst, YAML i Python; pliki binarne, takie jak NumPy (.npy, .npz), QPY i pickle; archiwa tar, gzip i zip	Pliki CSV, PNG, PDF, HDF5 i puste pliki

Nazwy plików

• Używaj tylko liter, cyfr, podkreśleń (_), myślników (-) i kropek (.). Zarządzany przepływ pracy usuwa z nazwy inne znaki, takie jak spacje, gdy pobierasz lub usuwasz plik. Na przykład możesz przesłać my results.json, ale nie możesz go pobrać.

• Magazyn plików nie ma folderów. file_upload() zachowuje tylko nazwę pliku, więc inputs/params.json jest zapisywany jako params.json.

Odczytywanie plików w przepływie pracy

Przekaż nazwę pliku podczas uruchamiania przepływu pracy:

run_workflow_remotely.ipynb

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

Aby to umożliwić, zmień kod przepływu pracy tak, aby odczytywał plik z /function_user_data, czyli miejsca, w którym każde zadanie widzi magazyn plików:

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

Wewnątrz zadania odczytuj i zapisuj pliki przez ten folder. Metody plikowe klienta, takie jak file_upload() i file_download(), służą do pracy w lokalnym notebooku.

Aby odczytać archiwum, wypakuj je na dysk lokalny:

source_files/my_workflow.py

import tarfile

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

Zapisuj pliki tymczasowe w /tmp. Magazyn plików to magazyn obiektów, który zadanie widzi jako folder, więc jest wolniejszy niż dysk lokalny. Wykonuj intensywne odczyty i zapisy na dysku lokalnym, a gotowe pliki kopiuj do magazynu plików.

Zapisz pliki wyjściowe

Twój lokalny notebook odczytuje nazwy plików wyjściowych z wyniku:

run_workflow_remotely.ipynb

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

Aby to umożliwić, zapisz każdy plik wyjściowy na najwyższym poziomie /function_user_data i zwróć nazwy plików za 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"]})

Pliki, które zadanie zapisuje poza /function_user_data, na przykład w /tmp lub w bieżącym folderze, są usuwane po zakończeniu zadania.

Pobierz pliki wyjściowe

Po zakończeniu zadania pobierz każdy plik za pomocą file_download(). Ustaw target_name, aby wybrać nazwę pliku lokalnego, oraz download_location, aby wybrać istniejący folder lokalny:

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 funkcja file_download() dodaje do nazwy pliku lokalnego losowy prefiks, na przykład downloaded_3f2a9c1b_energies.npy, i zwraca użytą nazwę.

Pobierz wiele wyników jako jedno archiwum

Jeśli zadanie zapisuje wiele plików wyjściowych lub pliki w podfolderach, spakuj je w przepływie pracy do jednego archiwum i pobierz tylko to archiwum:

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

Aby to umożliwić, zapisz wyniki w folderze na dysku lokalnym, dodaj folder do archiwum w /tmp i skopiuj archiwum do magazynu plików:

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

Ponieważ archiwum jest budowane na dysku lokalnym i kopiowane w jednym kroku, magazyn plików nigdy nie zawiera niekompletnego archiwum. Archiwum zachowuje strukturę folderów i może zawierać pliki dowolnego typu. W tym przykładzie pliki są wypakowywane do outputs/<run_name>/.

Usuń pliki

Zarządzany przepływ pracy nie usuwa plików z magazynu plików, więc pliki pozostają, dopóki nie usuniesz ich za pomocą file_delete():

run_workflow_remotely.ipynb

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

file_delete() zgłasza sukces nawet wtedy, gdy plik nie istnieje, więc najpierw sprawdź nazwę za pomocą files(). Aby usunąć wszystkie pliki jednego uruchomienia, przefiltruj files() według prefiksu nazwy uruchomienia, jak opisano w następnej sekcji:

run_workflow_remotely.ipynb

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

Dobra praktyka: nadaj plikom każdego uruchomienia unikalną nazwę

Wszystkie Twoje uruchomienia przepływu pracy korzystają z tego samego magazynu plików, a plik zapisany przez zadanie zastępuje każdy zapisany plik o tej samej nazwie. Jeśli uruchomisz kilka przebiegów jednocześnie i każdy zapisze energies.npy, zostanie zachowany tylko ostatni.

Jako dobrą praktykę przekaż nazwę uruchomienia jako argument i rozpoczynaj od niej nazwę każdego pliku zapisywanego przez to uruchomienie:

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)

Wszystkie uruchomienia mogą czytać ten sam plik wejściowy, hamiltonian.npz, ponieważ tylko go odczytują. Wspólne dane wejściowe prześlij jeden raz, przed rozpoczęciem uruchomień.

Używaj nazw uruchomień, które później rozpoznasz i które zawierają tylko litery, cyfry, podkreślenia, myślniki i kropki. Dzięki jednemu prefiksowi dla każdego uruchomienia możesz razem znajdować, pobierać i usuwać pliki danego uruchomienia.

Dobra praktyka: twórz punkty kontrolne w długich zadaniach

Zadanie, które zostanie anulowane, zakończy się błędem lub osiągnie limit 24 godzin, traci wszystko, czego nie ma w magazynie plików. W długich uruchomieniach zapisuj punkt kontrolny w magazynie plików w regularnych odstępach i wznawiaj od niego, gdy ponownie uruchamiasz przepływ pracy z tą samą nazwą uruchomienia:

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)

Aby kontynuować uruchomienie, które zostało zatrzymane, wywołaj ponownie run() z tymi samymi argumentami.

Pamiętaj o następujących kwestiach:

• Najpierw zapisz punkt kontrolny na dysku lokalnym, a potem go skopiuj. Jeśli zadanie zatrzyma się w trakcie zapisu, zapisany punkt kontrolny jest niekompletny i następne uruchomienie nie może go odczytać.

• Wybierz interwał dopasowany do kosztu jednego kroku. Zapisuj wystarczająco często, aby nie tracić dużo pracy, ale nie tak często, by zapisywanie punktów kontrolnych spowalniało zadanie.

• Usuń punkty kontrolne po udanym uruchomieniu. Kolejne uruchomienie o tej samej nazwie zaczyna się wtedy od początku.

W przypadku anulowanego zadania możesz też zapisać wyniki częściowe w obsłudze SIGTERM.

Następne kroki

Zalecenia

• Uruchom przeszukiwanie parametrów, aby rozpocząć jedno zadanie dla każdych danych wejściowych.
• Zapisuj logi i obsługuj błędy, aby zapisać wyniki częściowe po anulowaniu zadania.