実行をまたいでファイルを管理する

マネージドワークフローのジョブは、HPC の計算ノードで動くバッチジョブのように動作します。ジョブのローカルディスクは作業用の領域です。ジョブはそこに書き込めますが、ジョブの終了時にワークフローがそれを削除します。ジョブの終了後もファイルを保持するには、自分専用のストレージバケットのように機能するファイルストレージに保存します。アップロードしたファイルや、ワークフローが保存したファイルはファイルストレージに残り、ワークフローの以降の実行から読み取ることができます。

各ワークフローには、ユーザーごとに専用のファイルストレージがあります。同じワークフローを実行する他のユーザーに、あなたのファイルは表示されません。

入力ファイルをアップロードする

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

サポートされているファイルタイプ

マネージドワークフローは、アップロードされた各ファイルの拡張子ではなく内容を確認し、他のタイプのファイルを拒否します。

受け付けるもの	拒否されるもの
JSON、プレーンテキスト、YAML、Python ファイル、NumPy（.npy、.npz）、QPY、pickle ファイルなどのバイナリファイル、tar、gzip、zip アーカイブ	CSV、PNG、PDF、HDF5、空のファイル

ファイル名

• 英字、数字、アンダースコア（_）、ハイフン（-）、ピリオド（.）のみを使用してください。 ファイルをダウンロードまたは削除するとき、マネージドワークフローは、スペースなどのその他の文字を名前から削除します。たとえば、my results.json はアップロードできますが、ダウンロードはできません。

• ファイルストレージにはフォルダがありません。 file_upload() はファイル名だけを保持するため、inputs/params.json は params.json として保存されます。

ワークフローでファイルを読み取る

ワークフローを実行するときに、ファイルの名前を渡します。

run_workflow_remotely.ipynb

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

これを有効にするには、ワークフローのコードを編集して、/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）を付け、使用した名前を返します。

多数の出力を1つのアーカイブとしてダウンロードする

ジョブが多数の出力ファイルやサブフォルダ内のファイルを保存する場合は、ワークフロー内でそれらを1つのアーカイブにまとめ、そのアーカイブだけをダウンロードします。

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

アーカイブはローカルディスク上で作成され、1回の操作でコピーされるため、ファイルストレージに不完全なアーカイブが残ることはありません。アーカイブはフォルダ構造を保持し、任意のタイプのファイルを含めることができます。この例では、ファイルは outputs/<run_name>/ に展開されます。

ファイルを削除する

マネージドワークフローはファイルストレージからファイルを削除しないため、file_delete() で削除するまでファイルは残ります。

run_workflow_remotely.ipynb

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

file_delete() は、ファイルが存在しない場合でも成功を報告するため、先に files() で名前を確認してください。1回の実行のすべてのファイルを削除するには、次のセクションで説明するように、実行名のプレフィックスで files() をフィルタリングします。

run_workflow_remotely.ipynb

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

ベストプラクティス：各実行のファイルに一意の名前を付ける

ワークフローのすべての実行は同じファイルストレージを共有し、ジョブが保存したファイルは、同じ名前で保存済みのファイルを置き換えます。複数の実行を同時に開始し、それぞれが energies.npy を保存する場合、最後の1つだけが保持されます。

ベストプラクティスとして、実行名を引数として渡し、その実行が保存するすべてのファイルの名前をその実行名で始めてください。

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 を読み取ることができます。読み取るだけだからです。共有の入力は、実行を開始する前に一度だけアップロードしてください。

後から識別できる実行名を使い、文字、数字、アンダースコア、ハイフン、ピリオドのみを含めてください。実行ごとに1つのプレフィックスを付けておくと、1回の実行のファイルをまとめて検索、ダウンロード、削除できます。

ベストプラクティス: 長時間のジョブにはチェックポイントを設定する

キャンセルされたジョブ、失敗したジョブ、24時間の制限に達したジョブでは、ファイルストレージにないものはすべて失われます。長時間の実行では、一定の間隔でチェックポイントをファイルストレージに保存し、同じ実行名でワークフローを再度実行するときにそこから再開してください。

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() をもう一度呼び出します。

次の点に注意してください。

• チェックポイントはまずローカルディスクに書き込み、その後コピーしてください。 書き込み中にジョブが停止すると、保存されたチェックポイントが不完全になり、次の実行がそれを読み取れません。

• 1ステップのコストに合った間隔を選んでください。 失う作業が多くならない程度にこまめに保存し、ただしチェックポイントの書き込みでジョブが遅くならない程度の頻度にしてください。

• 実行が成功したらチェックポイントを削除してください。 そうすれば、同じ名前での次の実行は最初から始まります。

キャンセルされるジョブでは、SIGTERM ハンドラーで部分的な結果を保存することもできます。

次のステップ

おすすめ

• パラメーター・スイープを実行すると、入力ごとに1つのジョブを開始できます。
• ログを書き込み、エラーを処理すると、ジョブがキャンセルされたときに部分的な結果を保存できます。