إدارة الملفات عبر عمليات التشغيل

تعمل مهمة سير العمل المُدار مثل مهمة دفعية على عقدة حوسبة 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']

أنواع الملفات المدعومة

يفحص سير العمل المُدار محتويات كل ملف مرفوع، لا امتداده، ويرفض الملفات من الأنواع الأخرى:

Accepted	Rejected
JSON, plain text, YAML, and Python files; binary files such as NumPy (.npy, .npz), QPY, and pickle files; tar, gzip, and zip archives	CSV, PNG, PDF, HDF5, and empty files

أسماء الملفات

• استخدم الحروف والأرقام والشرطات السفلية (_) والواصلات (-) والنقاط (.) فقط. يزيل سير العمل المُدار الأحرف الأخرى، مثل المسافات، من الاسم عند تنزيل ملف أو حذفه. مثلًا، يمكنك رفع 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، وتُرجع الاسم الذي استخدمته.

تنزيل مخرجات كثيرة كأرشيف واحد

إذا كانت المهمة تحفظ ملفات إخراج كثيرة، أو ملفات في مجلدات فرعية، فاجمعها في أرشيف واحد داخل سير العمل، ونزّل الأرشيف فقط:

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.

الخطوات التالية

توصيات

• شغّل مسحًا للمعاملات لبدء مهمة واحدة لكل مدخل.
• اكتب السجلات وعالج الأخطاء لحفظ النتائج الجزئية عند إلغاء مهمة.