Pamahalaan ang mga file sa iba't ibang run

Ang managed workflow job ay gumagana tulad ng batch job sa isang HPC compute node. Ang local disk ng job ay scratch space. Maaaring magsulat dito ang job, pero binubura ito ng workflow kapag natapos ang job. Para panatilihin ang mga file pagkatapos matapos ang job, i-save ang mga ito sa file storage, na gumagana tulad ng sarili mong storage bucket. Ang mga file na ina-upload mo, at ang mga file na sine-save ng workflow mo, ay nananatili sa file storage, at mababasa ng mga susunod na run ng workflow ang mga ito.

Ang bawat workflow ay may sariling file storage para sa bawat user. Hindi nakikita ng ibang user na nagpapatakbo ng parehong workflow ang iyong mga file.

Mag-upload ng input file

Mag-upload ng file mula sa local notebook mo gamit ang file_upload(), at pagkatapos ay ilista ang mga file sa storage gamit ang 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']

Mga sinusuportahang uri ng file

Sinusuri ng managed workflow ang nilalaman ng bawat na-upload na file, hindi ang extension nito, at tinatanggihan ang mga file ng ibang uri:

Accepted	Rejected
JSON, plain text, YAML, and Python files; binary files such as NumPy (.npy, .npz), QPY, and pickle files; tar, gzip, and zip archives	CSV, PNG, PDF, HDF5, and empty files

Mga pangalan ng file

• Letra, digit, underscore (_), hyphen (-), at tuldok (.) lang ang gamitin. Inaalis ng managed workflow ang ibang karakter, tulad ng mga espasyo, sa pangalan kapag nag-download o nagbura ka ng file. Halimbawa, maaari mong i-upload ang my results.json, pero hindi mo ito ma-download.

• Walang folder ang file storage. Ang pangalan lang ng file ang itinatago ng file_upload(), kaya ang inputs/params.json ay sine-save bilang params.json.

Magbasa ng mga file sa workflow

Ipasa ang pangalan ng file kapag pinatakbo mo ang workflow:

run_workflow_remotely.ipynb

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

Para paganahin ito, i-edit ang code ng workflow mo para basahin ang file mula sa /function_user_data, kung saan nakikita ng bawat job ang file storage:

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

Sa loob ng job, magbasa at magsulat ng mga file sa pamamagitan ng folder na ito. Ang mga client file method, tulad ng file_upload() at file_download(), ay para sa local notebook mo.

Para magbasa ng archive, i-extract ito sa local disk:

source_files/my_workflow.py

import tarfile

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

Magsulat ng mga pansamantalang file sa /tmp. Ang file storage ay object storage na nakikita ng job bilang folder, kaya mas mabagal ang file storage kaysa sa local disk. Gawin ang mabibigat na pagbasa at pagsulat sa local disk, at kopyahin ang mga natapos na file sa file storage.

I-save ang mga output file

Binabasa ng local notebook mo ang mga pangalan ng output file mula sa resulta:

run_workflow_remotely.ipynb

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

Para paganahin ito, i-save ang bawat output file sa top level ng /function_user_data, at ibalik ang mga pangalan ng file gamit ang 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"]})

Ang mga file na isinusulat ng job sa labas ng /function_user_data, gaya ng sa /tmp o sa kasalukuyang folder, ay made-delete kapag natapos na ang job.

I-download ang mga output file

Pagkatapos matapos ang job, i-download ang bawat file gamit ang file_download(). Itakda ang target_name para piliin ang pangalan ng local na file, at ang download_location para piliin ang isang umiiral na local na folder:

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

Kapag walang target_name, nagdaragdag ang file_download() ng random na prefix sa pangalan ng local na file, gaya ng downloaded_3f2a9c1b_energies.npy, at ibinabalik nito ang pangalang ginamit nito.

I-download ang maraming output bilang isang archive

Kung nagse-save ang isang job ng maraming output file, o mga file sa mga subfolder, pagsama-samahin ang mga ito sa isang archive sa workflow, at ang archive lang ang i-download:

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 magawa ito, isulat ang mga output sa isang folder sa local na disk, idagdag ang folder sa isang archive sa /tmp, at kopyahin ang archive sa file storage:

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

Dahil ginagawa ang archive sa local na disk at kinokopya nang sabay-sabay, hindi kailanman magkakaroon ang file storage ng hindi kumpletong archive. Pinapanatili ng archive ang istruktura ng folder, at maaari itong maglaman ng mga file ng anumang uri. Sa halimbawang ito, ang mga file ay ine-extract sa outputs/<run_name>/.

I-delete ang mga file

Hindi nagde-delete ng mga file mula sa file storage ang isang managed workflow, kaya nananatili ang mga file hanggang i-delete mo ang mga ito gamit ang file_delete():

run_workflow_remotely.ipynb

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

Nagre-report ng tagumpay ang file_delete() kahit hindi umiiral ang file, kaya tingnan muna ang pangalan gamit ang files(). Para i-delete ang lahat ng file ng isang run, i-filter ang files() ayon sa prefix ng pangalan ng run, gaya ng inilalarawan sa susunod na seksyon:

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: bigyan ng natatanging pangalan ang mga file ng bawat run

Lahat ng run mo ng isang workflow ay nagbabahagi ng parehong file storage, at pinapalitan ng file na sine-save ng isang job ang anumang naka-store na file na may parehong pangalan. Kung magsisimula ka ng ilang run nang sabay at nagse-save ang bawat run ng energies.npy, ang huli lang ang mananatili.

Bilang best practice, magpasa ng pangalan ng run bilang argument, at simulan dito ang pangalan ng bawat file na sine-save ng run:

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)

Lahat ng run ay makakabasa ng parehong input file, hamiltonian.npz, dahil binabasa lang nila ito. Mag-upload ng shared na input nang isang beses, bago mo simulan ang mga run.

Gumamit ng mga pangalan ng run na makikilala mo sa hinaharap, at naglalaman lang ng mga letra, digit, underscore, hyphen, at tuldok. Sa isang prefix para sa bawat run, mahahanap, ma-di-download, at made-delete mo nang sabay ang mga file ng isang run.

Best practice: mag-checkpoint ng mahahabang job

Ang job na kinansela, pumalya, o umabot sa 24-oras na limitasyon ay nawawalan ng lahat ng wala sa file storage. Para sa mahahabang run, mag-save ng checkpoint sa file storage sa regular na pagitan, at magpatuloy mula rito kapag pinatakbo mo ulit ang workflow gamit ang parehong pangalan ng run:

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 ipagpatuloy ang run na tumigil, tawagin muli ang run() gamit ang parehong mga argument.

Tandaan ang mga sumusunod na punto:

• Isulat muna ang checkpoint sa local na disk, saka ito kopyahin. Kung tumigil ang job habang nagsusulat, hindi kumpleto ang naka-store na checkpoint, at hindi ito mababasa ng susunod na run.

• Pumili ng pagitan na tugma sa gastos ng isang hakbang. Mag-save nang sapat na dalas para hindi masyadong marami ang mawalang trabaho, pero hindi sobrang dalas na babagal ang job dahil sa pagsulat ng mga checkpoint.

• I-delete ang mga checkpoint pagkatapos magtagumpay ang run. Magsisimula ulit sa umpisa ang susunod na run na may parehong pangalan.

Para sa job na kinansela, maaari ka ring mag-save ng mga partial na resulta sa isang SIGTERM handler.

Mga susunod na hakbang

Mga rekomendasyon

• Magpatakbo ng parameter sweep para magsimula ng isang job para sa bawat input.
• Magsulat ng mga log at humawak ng mga error para mag-save ng mga partial na resulta kapag kinansela ang isang job.