จัดการไฟล์ข้ามการรัน

job ของ managed workflow ทำงานเหมือน batch job บนโหนดประมวลผล HPC ดิสก์ภายในของ job เป็นพื้นที่ชั่วคราว job เขียนลงไปได้ แต่ workflow จะลบทิ้งเมื่อ job จบ หากต้องการเก็บไฟล์ไว้หลัง job จบ ให้บันทึกลงที่เก็บไฟล์ (file storage) ซึ่งทำงานเหมือนบักเก็ตเก็บข้อมูลของคุณเอง ไฟล์ที่คุณอัปโหลดและไฟล์ที่ 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']

ชนิดไฟล์ที่รองรับ

managed workflow ตรวจเนื้อหาของแต่ละไฟล์ที่อัปโหลด ไม่ใช่นามสกุล และปฏิเสธไฟล์ชนิดอื่น

Accepted	Rejected
ไฟล์ JSON, ข้อความธรรมดา, YAML และ Python; ไฟล์ไบนารี เช่น NumPy (.npy, .npz), QPY และ pickle; ไฟล์เก็บถาวร tar, gzip และ zip	ไฟล์ CSV, PNG, PDF, HDF5 และไฟล์ว่าง

ชื่อไฟล์

• ใช้เฉพาะตัวอักษร ตัวเลข ขีดล่าง (_) ยัติภังค์ (-) และจุด (.) managed 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 ซึ่งเป็นที่ที่ทุก job เห็นที่เก็บไฟล์

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

ภายใน job ให้อ่านและเขียนไฟล์ผ่านโฟลเดอร์นี้ เมธอดไฟล์ของไคลเอนต์ เช่น 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 ที่เก็บไฟล์เป็น object storage ที่ job เห็นเป็นโฟลเดอร์ ที่เก็บไฟล์จึงช้ากว่าดิสก์ภายใน ให้อ่านและเขียนหนักๆ บนดิสก์ภายใน แล้วคัดลอกไฟล์ที่เสร็จแล้วไปยังที่เก็บไฟล์

บันทึกไฟล์เอาต์พุต

โน้ตบุ๊กในเครื่องของคุณอ่านชื่อไฟล์เอาต์พุตจากผลลัพธ์

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

ไฟล์ที่ job เขียนนอก /function_user_data เช่นใน /tmp หรือโฟลเดอร์ปัจจุบัน จะถูกลบเมื่อ job สิ้นสุด

ดาวน์โหลดไฟล์ผลลัพธ์

หลังจาก job สิ้นสุด ให้ดาวน์โหลดแต่ละไฟล์ด้วย 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 และคืนชื่อที่ใช้จริงกลับมา

ดาวน์โหลดผลลัพธ์จำนวนมากเป็นไฟล์เก็บถาวรไฟล์เดียว

ถ้า job บันทึกไฟล์ผลลัพธ์จำนวนมาก หรือไฟล์ในโฟลเดอร์ย่อย ให้รวมไฟล์เหล่านั้นเป็นไฟล์เก็บถาวรไฟล์เดียวใน 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>/

ลบไฟล์

managed 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 ทั้งหมดของคุณใช้ที่เก็บไฟล์เดียวกัน และไฟล์ที่ job บันทึกจะแทนที่ไฟล์ที่เก็บไว้ซึ่งมีชื่อเดียวกัน ถ้าคุณเริ่มหลายการรันพร้อมกันและแต่ละการรันบันทึก 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 ได้ เพราะอ่านอย่างเดียว ให้อัปโหลดอินพุตที่ใช้ร่วมกันครั้งเดียวก่อนเริ่มการรัน

ใช้ชื่อการรันที่คุณจำได้ภายหลัง และมีเฉพาะตัวอักษร ตัวเลข เครื่องหมายขีดล่าง ยัติภังค์ และจุด เมื่อมีคำนำหน้าหนึ่งคำต่อหนึ่งการรัน คุณจะค้นหา ดาวน์โหลด และลบไฟล์ของการรันนั้นพร้อมกันได้

แนวปฏิบัติที่ดี: สร้าง checkpoint สำหรับ job ที่รันนาน

job ที่ถูกยกเลิก ล้มเหลว หรือถึงขีดจำกัด 24 ชั่วโมง จะสูญเสียทุกอย่างที่ไม่ได้อยู่ในที่เก็บไฟล์ สำหรับการรันที่ใช้เวลานาน ให้บันทึก checkpoint ลงที่เก็บไฟล์เป็นระยะ และกลับมาทำต่อจาก checkpoint นั้นเมื่อรัน 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() อีกครั้งด้วยอาร์กิวเมนต์เดิม

โปรดคำนึงถึงประเด็นต่อไปนี้:

• เขียน checkpoint ลงดิสก์ในเครื่องก่อน แล้วค่อยคัดลอก ถ้า job หยุดระหว่างเขียน checkpoint ที่เก็บไว้จะไม่สมบูรณ์ และการรันครั้งถัดไปจะอ่านไม่ได้

• เลือกช่วงเวลาที่เหมาะกับต้นทุนของแต่ละขั้นตอน บันทึกให้บ่อยพอที่จะไม่เสียงานมาก แต่ไม่บ่อยเกินไปจนการเขียน checkpoint ทำให้ job ช้าลง

• ลบ checkpoint หลังจากการรันสำเร็จ การรันครั้งต่อมาที่ใช้ชื่อเดียวกันจะเริ่มจากต้น

สำหรับ job ที่ถูกยกเลิก คุณยังบันทึกผลลัพธ์บางส่วนใน SIGTERM handler ได้ด้วย

ขั้นตอนถัดไป

คำแนะนำ

• รัน parameter sweep เพื่อเริ่มหนึ่ง job ต่อหนึ่งอินพุต
• เขียน log และจัดการข้อผิดพลาด เพื่อบันทึกผลลัพธ์บางส่วนเมื่อ job ถูกยกเลิก