Urus fail merentas larian

Tugas aliran kerja terurus berfungsi seperti tugas kelompok pada nod pengiraan HPC. Cakera setempat tugas ialah ruang calar. Tugas boleh menulis padanya, tetapi aliran kerja memadamnya apabila tugas tamat. Untuk mengekalkan fail selepas tugas tamat, simpannya ke storan fail, yang berfungsi seperti baldi storan anda sendiri. Fail yang anda muat naik, dan fail yang disimpan oleh aliran kerja anda, kekal dalam storan fail, dan larian aliran kerja kemudian boleh membacanya.

Setiap aliran kerja mempunyai storan failnya sendiri bagi setiap pengguna. Pengguna lain yang menjalankan aliran kerja yang sama tidak melihat fail anda.

Muat naik fail input

Muat naik fail daripada buku nota setempat anda dengan file_upload(), kemudian senaraikan fail dalam storan 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']

Jenis fail yang disokong

Aliran kerja terurus menyemak kandungan setiap fail yang dimuat naik, bukan sambungannya, dan menolak fail jenis lain:

Diterima	Ditolak
Fail JSON, teks biasa, YAML dan Python; fail binari seperti NumPy (.npy, .npz), QPY dan pickle; arkib tar, gzip dan zip	Fail CSV, PNG, PDF, HDF5 dan fail kosong

Nama fail

• Gunakan hanya huruf, digit, garis bawah (_), tanda sempang (-), dan noktah (.). Aliran kerja terurus membuang aksara lain, seperti ruang, daripada nama apabila anda memuat turun atau memadam fail. Contohnya, anda boleh memuat naik my results.json, tetapi anda tidak boleh memuat turunnya.

• Storan fail tidak mempunyai folder. file_upload() hanya mengekalkan nama fail, jadi inputs/params.json disimpan sebagai params.json.

Baca fail dalam aliran kerja

Hantar nama fail apabila anda menjalankan aliran kerja:

run_workflow_remotely.ipynb

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

Untuk membolehkan ini, edit kod aliran kerja anda untuk membaca fail daripada /function_user_data, iaitu tempat setiap tugas melihat storan fail:

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 tugas, baca dan tulis fail melalui folder ini. Kaedah fail klien, seperti file_upload() dan file_download(), adalah untuk buku nota setempat anda.

Untuk membaca arkib, ekstraknya ke cakera setempat:

source_files/my_workflow.py

import tarfile

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

Tulis fail sementara ke /tmp. Storan fail ialah storan objek yang dilihat oleh tugas sebagai folder, jadi storan fail lebih perlahan daripada cakera setempat. Lakukan bacaan dan tulisan yang berat pada cakera setempat, dan salin fail yang siap ke storan fail.

Simpan fail output

Buku nota setempat anda membaca nama fail output daripada hasil:

run_workflow_remotely.ipynb

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

Untuk membolehkan ini, simpan setiap fail output pada peringkat teratas /function_user_data, dan kembalikan nama fail 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"]})

Fail yang ditulis oleh tugas di luar /function_user_data, seperti dalam /tmp atau folder semasa, dipadamkan apabila tugas tamat.

Muat turun fail output

Selepas tugas tamat, muat turun setiap fail dengan file_download(). Tetapkan target_name untuk memilih nama fail setempat, dan download_location untuk memilih folder setempat sedia 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() menambah awalan rawak pada nama fail setempat, seperti downloaded_3f2a9c1b_energies.npy, dan mengembalikan nama yang digunakannya.

Muat turun banyak output sebagai satu arkib

Jika tugas menyimpan banyak fail output, atau fail dalam subfolder, himpunkannya ke dalam satu arkib dalam aliran kerja, dan muat turun arkib sahaja:

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

Untuk membolehkan ini, tulis output ke folder pada cakera setempat, tambahkan folder itu ke arkib dalam /tmp, dan salin arkib ke storan fail:

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

Oleh sebab arkib dibina pada cakera setempat dan disalin dalam satu langkah, storan fail tidak pernah mengandungi arkib yang tidak lengkap. Arkib mengekalkan struktur folder, dan ia boleh mengandungi fail jenis apa pun. Dalam contoh ini, fail diekstrak ke outputs/<run_name>/.

Padam fail

Aliran kerja terurus tidak memadam fail daripada storan fail, jadi fail kekal sehingga anda memadamnya dengan file_delete():

run_workflow_remotely.ipynb

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

file_delete() melaporkan kejayaan walaupun fail tidak wujud, jadi semak nama dengan files() terlebih dahulu. Untuk memadam semua fail satu larian, tapis files() mengikut awalan nama larian, seperti yang diterangkan dalam bahagian seterusnya:

run_workflow_remotely.ipynb

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

Amalan terbaik: beri setiap fail larian nama yang unik

Semua larian aliran kerja anda berkongsi storan fail yang sama, dan fail yang disimpan oleh sesuatu job menggantikan mana-mana fail tersimpan yang bernama sama. Jika anda memulakan beberapa larian serentak dan setiap larian menyimpan energies.npy, hanya yang terakhir dikekalkan.

Sebagai amalan terbaik, hantar nama larian sebagai argumen, dan mulakan nama setiap fail yang disimpan oleh larian itu dengannya:

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 larian boleh membaca fail input yang sama, hamiltonian.npz, kerana mereka hanya membacanya. Muat naik input kongsi sekali sahaja, sebelum anda memulakan larian.

Gunakan nama larian yang boleh anda kenali kemudian, dan yang hanya mengandungi huruf, digit, garis bawah, tanda sempang dan noktah. Dengan satu awalan bagi setiap larian, anda boleh mencari, memuat turun dan memadam fail sesuatu larian bersama-sama.

Amalan terbaik: buat titik semak untuk job yang panjang

Job yang dibatalkan, gagal atau mencapai had 24 jam kehilangan segala-galanya yang tidak berada dalam storan fail. Untuk larian yang panjang, simpan titik semak ke storan fail pada sela masa tetap, dan sambung semula daripadanya apabila anda menjalankan aliran kerja sekali lagi dengan nama larian 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 menyambung larian yang telah berhenti, panggil run() sekali lagi dengan argumen yang sama.

Ingat perkara berikut:

• Tulis titik semak ke cakera setempat dahulu, kemudian salinnya. Jika job berhenti semasa penulisan, titik semak yang disimpan tidak lengkap, dan larian seterusnya tidak dapat membacanya.

• Pilih sela masa yang sepadan dengan kos satu langkah. Simpan cukup kerap supaya anda tidak kehilangan banyak kerja, tetapi tidak terlalu kerap sehingga penulisan titik semak melambatkan job.

• Padam titik semak selepas larian berjaya. Larian kemudian dengan nama yang sama akan bermula dari awal.

Bagi job yang dibatalkan, anda juga boleh menyimpan keputusan separa dalam pengendali SIGTERM.

Langkah seterusnya

Cadangan

• Jalankan penyapuan parameter untuk memulakan satu job bagi setiap input.
• Tulis log dan kendalikan ralat untuk menyimpan keputusan separa apabila job dibatalkan.