Manage files across runs

A managed workflow job works like a batch job on an HPC compute node. The job's local disk is scratch space. The job can write to it, but the workflow deletes it when the job ends. To keep files after the job ends, save them to file storage, which works like your own storage bucket. Files that you upload, and files that your workflow saves, remain in file storage, and later runs of the workflow can read them.

Each workflow has its own file storage for each user. Other users who run the same workflow do not see your files.

Upload an input file

Upload a file from your local notebook with file_upload(), and then list the files in storage with files():

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']

Supported file types

A managed workflow checks the contents of each uploaded file, not its extension, and rejects files of other types:

File names

• Use only letters, digits, underscores (_), hyphens (-), and periods (.). A managed workflow removes other characters, such as spaces, from the name when you download or delete a file. For example, you can upload my results.json, but you cannot download it.
• File storage has no folders. file_upload() keeps only the file name, so inputs/params.json is stored as params.json.

Read files in a workflow

Pass the name of the file when you run the workflow:

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

To enable this, edit your workflow code to read the file from /function_user_data, which is where every job sees file storage:

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

Inside a job, read and write files through this folder. The client file methods, such as file_upload() and file_download(), are for your local notebook.

To read an archive, extract it to the local disk:

import tarfile

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

Write temporary files to /tmp. File storage is object storage that the job sees as a folder, so file storage is slower than the local disk. Do heavy reads and writes on the local disk, and copy finished files to file storage.

Save output files

Your local notebook reads the names of the output files from the result:

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

To enable this, save each output file at the top level of /function_user_data, and return the file names with save_result():

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

Files that the job writes outside /function_user_data, such as in /tmp or the current folder, are deleted when the job ends.

Download output files

After the job ends, download each file with file_download(). Set target_name to choose the local file name, and download_location to choose an existing local folder:

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

Without target_name, file_download() adds a random prefix to the local file name, such as downloaded_3f2a9c1b_energies.npy, and returns the name that it used.

Download many outputs as one archive

If a job saves many output files, or files in subfolders, bundle them into one archive in the workflow, and download only the archive:

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

To enable this, write the outputs to a folder on the local disk, add the folder to an archive in /tmp, and copy the archive to file storage:

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

Because the archive is built on the local disk and copied in one step, file storage never holds an incomplete archive. The archive keeps the folder structure, and it can contain files of any type. In this example, the files are extracted to outputs/<run_name>/.

Delete files

A managed workflow does not delete files from file storage, so files remain until you delete them with file_delete():

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

file_delete() reports success even if the file does not exist, so check the name with files() first. To delete all the files of one run, filter files() by the run's name prefix, as described in the next section:

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

Best practice: give each run's files a unique name

All of your runs of a workflow share the same file storage, and a file that a job saves replaces any stored file with the same name. If you start several runs at the same time and each run saves energies.npy, only the last one is kept.

As a best practice, pass a run name as an argument, and start the name of every file that the run saves with it:

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

run_name = arguments["run_name"]
np.save(FILE_STORAGE / f"{run_name}_energies.npy", energies)

All runs can read the same input file, hamiltonian.npz, because they only read it. Upload a shared input one time, before you start the runs.

Use run names that you can recognize later, and that contain only letters, digits, underscores, hyphens, and periods. With one prefix for each run, you can find, download, and delete the files of a run together.

Best practice: checkpoint long jobs

A job that is canceled, fails, or reaches the 24-hour limit loses everything that is not in file storage. For long runs, save a checkpoint to file storage at regular intervals, and resume from it when you run the workflow again with the same run name:

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)

To continue a run that stopped, call run() again with the same arguments.

Keep the following points in mind:

• Write the checkpoint to the local disk first, and then copy it. If the job stops during a write, the stored checkpoint is incomplete, and the next run cannot read it.
• Choose an interval that matches the cost of a step. Save often enough that you do not lose much work, but not so often that writing checkpoints slows the job.
• Delete checkpoints after the run succeeds. A later run with the same name then starts from the beginning.

For a job that is canceled, you can also save partial results in a SIGTERM handler.

Next steps