Kelola file lintas run

Job managed workflow bekerja seperti batch job di compute node HPC. Disk lokal job adalah ruang scratch. Job bisa menulis ke sana, tetapi workflow menghapusnya saat job berakhir. Untuk menyimpan file setelah job berakhir, simpan ke file storage, yang bekerja seperti bucket penyimpanan milikmu sendiri. File yang kamu unggah, dan file yang disimpan workflow-mu, tetap ada di file storage, dan run workflow berikutnya bisa membacanya.

Setiap workflow punya file storage sendiri untuk setiap pengguna. Pengguna lain yang menjalankan workflow yang sama tidak melihat file kamu.

Unggah file input

Unggah file dari notebook lokalmu dengan file_upload(), lalu daftar file di storage dengan 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']

Tipe file yang didukung

Managed workflow memeriksa isi setiap file yang diunggah, bukan ekstensinya, dan menolak file dengan tipe lain:

Diterima	Ditolak
File JSON, teks biasa, YAML, dan Python; file biner seperti NumPy (.npy, .npz), QPY, dan pickle; arsip tar, gzip, dan zip	File CSV, PNG, PDF, HDF5, dan file kosong

Nama file

• Pakai hanya huruf, angka, underscore (_), tanda hubung (-), dan titik (.). Managed workflow menghapus karakter lain, seperti spasi, dari nama saat kamu mengunduh atau menghapus file. Misalnya, kamu bisa mengunggah my results.json, tetapi tidak bisa mengunduhnya.

• File storage tidak punya folder. file_upload() hanya menyimpan nama file, jadi inputs/params.json disimpan sebagai params.json.

Baca file di workflow

Berikan nama file saat kamu menjalankan workflow:

run_workflow_remotely.ipynb

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

Supaya ini bisa jalan, ubah kode workflow-mu untuk membaca file dari /function_user_data, yaitu tempat setiap job melihat 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"])

Di dalam job, baca dan tulis file lewat folder ini. Method file klien, seperti file_upload() dan file_download(), ditujukan untuk notebook lokalmu.

Untuk membaca arsip, ekstrak ke disk lokal:

source_files/my_workflow.py

import tarfile

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

Tulis file sementara ke /tmp. File storage adalah object storage yang dilihat job sebagai folder, jadi file storage lebih lambat daripada disk lokal. Lakukan pembacaan dan penulisan yang berat di disk lokal, lalu salin file yang sudah selesai ke file storage.

Simpan file output

Notebook lokalmu membaca nama file output dari hasilnya:

run_workflow_remotely.ipynb

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

Supaya ini bisa jalan, simpan setiap file output di level teratas /function_user_data, dan kembalikan nama file dengan 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"]})

File yang ditulis job di luar /function_user_data, seperti di /tmp atau folder saat ini, dihapus saat job berakhir.

Unduh file output

Setelah job berakhir, unduh setiap file dengan file_download(). Atur target_name untuk memilih nama file lokal, dan download_location untuk memilih folder lokal yang sudah ada:

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

Tanpa target_name, file_download() menambahkan prefiks acak ke nama file lokal, seperti downloaded_3f2a9c1b_energies.npy, dan mengembalikan nama yang dipakainya.

Unduh banyak output sebagai satu arsip

Kalau job menyimpan banyak file output, atau file di subfolder, kemas semuanya menjadi satu arsip di workflow, dan unduh arsipnya saja:

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

Supaya ini bisa jalan, tulis output ke folder di disk lokal, tambahkan folder itu ke arsip di /tmp, lalu salin arsipnya ke 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})

Karena arsip dibuat di disk lokal dan disalin dalam satu langkah, file storage tidak pernah menyimpan arsip yang belum lengkap. Arsip mempertahankan struktur folder, dan bisa berisi file dengan tipe apa pun. Dalam contoh ini, file diekstrak ke outputs/<run_name>/.

Hapus file

Managed workflow tidak menghapus file dari file storage, jadi file tetap ada sampai kamu menghapusnya dengan file_delete():

run_workflow_remotely.ipynb

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

file_delete() melaporkan berhasil meskipun file-nya tidak ada, jadi cek dulu namanya dengan files(). Untuk menghapus semua file dari satu run, filter files() berdasarkan awalan nama run, seperti dijelaskan di bagian berikutnya:

run_workflow_remotely.ipynb

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

Praktik terbaik: beri setiap file run nama yang unik

Semua run dari satu workflow memakai penyimpanan file yang sama, dan file yang disimpan oleh sebuah job akan menggantikan file tersimpan mana pun dengan nama yang sama. Kalau kamu memulai beberapa run sekaligus dan tiap run menyimpan energies.npy, hanya yang terakhir yang tersimpan.

Sebagai praktik terbaik, kirim nama run sebagai argumen, dan awali nama setiap file yang disimpan run itu dengan nama tersebut:

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)

Semua run bisa membaca file input yang sama, hamiltonian.npz, karena mereka hanya membacanya. Unggah input bersama satu kali saja, sebelum kamu memulai run.

Pakai nama run yang mudah kamu kenali nanti, dan hanya berisi huruf, angka, garis bawah, tanda hubung, dan titik. Dengan satu awalan untuk tiap run, kamu bisa menemukan, mengunduh, dan menghapus file dari satu run sekaligus.

Praktik terbaik: buat checkpoint untuk job yang panjang

Job yang dibatalkan, gagal, atau mencapai batas 24 jam akan kehilangan semua yang tidak ada di penyimpanan file. Untuk run yang panjang, simpan checkpoint ke penyimpanan file secara berkala, lalu lanjutkan dari situ saat kamu menjalankan workflow lagi dengan nama run yang sama:

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)

Untuk melanjutkan run yang berhenti, panggil run() lagi dengan argumen yang sama.

Perhatikan poin-poin berikut:

• Tulis checkpoint ke disk lokal dulu, baru salin. Kalau job berhenti saat menulis, checkpoint yang tersimpan jadi tidak lengkap, dan run berikutnya tidak bisa membacanya.

• Pilih interval yang sesuai dengan biaya satu langkah. Simpan cukup sering supaya tidak banyak pekerjaan yang hilang, tapi jangan terlalu sering sampai penulisan checkpoint memperlambat job.

• Hapus checkpoint setelah run berhasil. Run berikutnya dengan nama yang sama akan mulai dari awal.

Untuk job yang dibatalkan, kamu juga bisa menyimpan hasil parsial di handler SIGTERM.

Langkah berikutnya

Rekomendasi

• Jalankan parameter sweep untuk memulai satu job untuk setiap input.
• Tulis log dan tangani error untuk menyimpan hasil parsial saat job dibatalkan.