실행 간 파일 관리

관리형 워크플로 작업은 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"]})

작업이 /tmp나 현재 폴더 등 /function_user_data 밖에 쓴 파일은 작업이 끝나면 삭제됩니다.

출력 파일 다운로드

작업이 끝난 후 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처럼 임의의 접두사를 붙이고 사용한 이름을 반환합니다.

많은 출력을 하나의 아카이브로 다운로드

작업이 많은 출력 파일이나 하위 폴더의 파일을 저장한다면 워크플로에서 이를 하나의 아카이브로 묶고 아카이브만 다운로드하세요.

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>/에 압축 해제됩니다.

파일 삭제

관리형 워크플로는 파일 스토리지의 파일을 삭제하지 않으므로, 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)

모범 사례: 각 실행의 파일에 고유한 이름 지정

한 워크플로의 모든 실행은 같은 파일 스토리지를 공유하며, 작업이 저장한 파일은 같은 이름으로 저장된 파일을 대체합니다. 여러 실행을 동시에 시작하고 각 실행이 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시간 제한에 도달한 작업은 파일 스토리지에 없는 모든 것을 잃습니다. 긴 실행에서는 일정한 간격으로 체크포인트를 파일 스토리지에 저장하고, 같은 실행 이름으로 워크플로를 다시 실행할 때 거기서부터 재개하세요.

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 handler에서 부분 결과를 저장할 수도 있습니다.

다음 단계

권장 사항

• 매개변수 스윕 실행으로 입력마다 작업을 하나씩 시작합니다.
• 로그 작성 및 오류 처리로 작업이 취소될 때 부분 결과를 저장합니다.