Керування файлами між запусками

Завдання керованого workflow працює як пакетне завдання на обчислювальному вузлі HPC. Локальний диск завдання є тимчасовим простором. Завдання може записувати на нього, але workflow видаляє його після завершення завдання. Щоб зберегти файли після завершення завдання, збережи їх у файлове сховище, яке працює як твій власний bucket. Файли, які ти завантажуєш, і файли, які зберігає твій workflow, залишаються у файловому сховищі, і наступні запуски workflow можуть їх читати.

Кожен workflow має власне файлове сховище для кожного користувача. Інші користувачі, які запускають той самий workflow, не бачать твоїх файлів.

Завантаж вхідний файл

Завантаж файл зі свого локального ноутбука за допомогою file_upload(), а потім виведи список файлів у сховищі за допомогою 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']

Підтримувані типи файлів

Керований workflow перевіряє вміст кожного завантаженого файлу, а не його розширення, і відхиляє файли інших типів:

Приймаються	Відхиляються
Файли JSON, звичайного тексту, YAML і Python; двійкові файли, як-от NumPy (.npy, .npz), QPY і pickle; архіви tar, gzip і zip	CSV, PNG, PDF, HDF5 і порожні файли

Імена файлів

• Використовуй лише літери, цифри, підкреслення (_), дефіси (-) і крапки (.). Керований workflow вилучає з імені інші символи, наприклад пробіли, коли ти завантажуєш або видаляєш файл. Наприклад, ти можеш завантажити my results.json, але не зможеш його скачати.

• Файлове сховище не має папок. file_upload() зберігає лише ім'я файлу, тому inputs/params.json зберігається як params.json.

Читання файлів у workflow

Передай ім'я файлу під час запуску workflow:

run_workflow_remotely.ipynb

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

Щоб це запрацювало, змін код свого workflow так, щоб він читав файл із /function_user_data — саме там кожне завдання бачить файлове сховище:

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

Усередині завдання читай і записуй файли через цю папку. Клієнтські методи роботи з файлами, як-от file_upload() і file_download(), призначені для твого локального ноутбука.

Щоб прочитати архів, розпакуй його на локальний диск:

source_files/my_workflow.py

import tarfile

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

Записуй тимчасові файли в /tmp. Файлове сховище — це об'єктне сховище, яке завдання бачить як папку, тому воно повільніше за локальний диск. Виконуй інтенсивне читання й запис на локальному диску, а готові файли копіюй у файлове сховище.

Збережи вихідні файли

Твій локальний ноутбук читає імена вихідних файлів із результату:

run_workflow_remotely.ipynb

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

Щоб це запрацювало, збережи кожен вихідний файл на верхньому рівні /function_user_data і поверни імена файлів за допомогою 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"]})

Файли, які завдання записує поза /function_user_data, наприклад у /tmp або поточну папку, видаляються після завершення завдання.

Скачай вихідні файли

Після завершення завдання скачай кожен файл за допомогою file_download(). Задай target_name, щоб вибрати локальне ім'я файлу, і download_location, щоб вибрати наявну локальну папку:

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

Без target_name функція file_download() додає до локального імені файлу випадковий префікс, наприклад downloaded_3f2a9c1b_energies.npy, і повертає використане ім'я.

Скачування багатьох результатів одним архівом

Якщо завдання зберігає багато вихідних файлів або файли в підпапках, об'єднай їх в один архів у workflow і скачай лише архів:

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

Щоб це запрацювало, запиши результати в папку на локальному диску, додай папку до архіву в /tmp і скопіюй архів у файлове сховище:

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

Оскільки архів створюється на локальному диску й копіюється одним кроком, файлове сховище ніколи не містить неповного архіву. Архів зберігає структуру папок і може містити файли будь-якого типу. У цьому прикладі файли розпаковуються в outputs/<run_name>/.

Видалення файлів

Керований workflow не видаляє файли з файлового сховища, тому файли залишаються, доки ти не видалиш їх за допомогою file_delete():

run_workflow_remotely.ipynb

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

file_delete() повідомляє про успіх, навіть якщо файлу не існує, тому спершу перевір ім'я за допомогою files(). Щоб видалити всі файли одного запуску, відфільтруй files() за префіксом імені запуску, як описано в наступному розділі:

run_workflow_remotely.ipynb

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

Рекомендація: давай файлам кожного запуску унікальні імена

Усі твої запуски workflow спільно використовують одне файлове сховище, а файл, який зберігає завдання, замінює будь-який збережений файл з таким самим іменем. Якщо ти запускаєш кілька запусків одночасно і кожен зберігає energies.npy, залишиться лише останній.

Як рекомендація, передавай ім'я запуску як аргумент і починай ним ім'я кожного файлу, який зберігає запуск:

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)

Усі запуски можуть читати той самий вхідний файл hamiltonian.npz, бо вони його лише читають. Завантажуй спільний вхідний файл один раз, перш ніж запускати запуски.

Використовуй імена запусків, які зможеш упізнати пізніше і які містять лише літери, цифри, підкреслення, дефіси й крапки. З одним префіксом для кожного запуску ти можеш разом знаходити, скачувати й видаляти файли запуску.

Рекомендація: роби контрольні точки в довгих завданнях

Завдання, яке скасовано, яке завершилося помилкою або досягло 24-годинного ліміту, втрачає все, чого немає у файловому сховищі. Для довгих запусків регулярно зберігай контрольну точку у файлове сховище й відновлюйся з неї, коли знову запускаєш workflow з тим самим іменем запуску:

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)

Щоб продовжити зупинений запуск, знову викличи run() з тими самими аргументами.

Май на увазі таке:

• Спершу запиши контрольну точку на локальний диск, а потім скопіюй її. Якщо завдання зупиниться під час запису, збережена контрольна точка буде неповною, і наступний запуск не зможе її прочитати.

• Вибери інтервал, що відповідає вартості кроку. Зберігай достатньо часто, щоб не втратити багато роботи, але не так часто, щоб запис контрольних точок сповільнював завдання.

• Видаляй контрольні точки після успішного завершення запуску. Тоді наступний запуск з тим самим іменем почнеться з початку.

Для скасованого завдання ти також можеш зберегти часткові результати в обробнику SIGTERM.

Наступні кроки

Рекомендації

• Запусти перебір параметрів, щоб запустити одне завдання для кожного входу.
• Записуй логи й обробляй помилки, щоб зберігати часткові результати, коли завдання скасовано.