ניהול קבצים בין הרצות

משימה בתהליך עבודה מנוהל פועלת כמו משימת batch על צומת חישוב של 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"]})

קבצים שה-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 ב-jobs ארוכים

job שבוטל, נכשל או הגיע למגבלה של 24 שעות מאבד כל מה שלא נמצא באחסון הקבצים. בהרצות ארוכות, שמרו 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 השמור חלקי, וההרצה הבאה לא תוכל לקרוא אותו.

• בחרו מרווח שמתאים לעלות של צעד. שמרו מספיק לעתים כדי לא לאבד הרבה עבודה, אבל לא כל כך לעתים שכתיבת ה-checkpoints תאט את ה-job.

• מחקו checkpoints אחרי שההרצה הצליחה. הרצה מאוחרת יותר עם אותו שם תתחיל אז מההתחלה.

ב-job שמבוטל, אפשר גם לשמור תוצאות חלקיות ב-handler של SIGTERM.

השלבים הבאים

המלצות

• הריצו סריקת פרמטרים כדי להתחיל job אחד לכל קלט.
• כתבו logs וטפלו בשגיאות כדי לשמור תוצאות חלקיות כש-job מבוטל.