Zum Hauptinhalt springen

Qiskit Serverless Rechen- und Datenressourcen verwalten

Paketversionen

Der Code auf dieser Seite wurde mit den folgenden Anforderungen entwickelt. Wir empfehlen, diese Versionen oder neuere zu verwenden.

qiskit[all]~=2.0.0
qiskit-ibm-runtime~=0.37.0
qiskit-serverless~=0.22.0

Mit Qiskit Serverless kannst du Rechen- und Datenressourcen in deinem Qiskit-Muster verwalten, einschließlich CPUs, QPUs und anderer Rechenbeschleuniger.

Detaillierte Status setzen​

Serverless-Workloads durchlaufen mehrere Phasen innerhalb eines Workflows. Standardmäßig sind die folgenden Status mit job.status() einsehbar:

  • QUEUED: Der Workload wartet auf klassische Ressourcen
  • INITIALIZING: Der Workload wird eingerichtet
  • RUNNING: Der Workload läuft derzeit auf klassischen Ressourcen
  • DONE: Der Workload wurde erfolgreich abgeschlossen

Du kannst außerdem benutzerdefinierte Status setzen, die die jeweilige Workflow-Phase genauer beschreiben, wie nachfolgend gezeigt.

# Added by doQumentation — required packages for this notebook
!pip install -q qiskit qiskit-ibm-runtime qiskit-serverless
# This cell is hidden from users, it just creates a new folder
from pathlib import Path

Path("./source_files").mkdir(exist_ok=True)
%%writefile ./source_files/status_example.py

from qiskit_serverless import update_status, Job

## If your function has a mapping stage, particularly application functions, you can set the status to "RUNNING: MAPPING" as follows:
update_status(Job.MAPPING)

## While handling transpilation, error suppression, and so forth, you can set the status to "RUNNING: OPTIMIZING_FOR_HARDWARE":
update_status(Job.OPTIMIZING_HARDWARE)

## After you submit jobs to Qiskit Runtime, the underlying quantum job will be queued. You can set status to "RUNNING: WAITING_FOR_QPU":
update_status(Job.WAITING_QPU)

## When the Qiskit Runtime job starts running on the QPU, set the following status "RUNNING: EXECUTING_QPU":
update_status(Job.EXECUTING_QPU)

## Once QPU is completed and post-processing has begun, set the status "RUNNING: POST_PROCESSING":
update_status(Job.POST_PROCESSING)
Writing ./source_files/status_example.py

Nach erfolgreichem Abschluss des Workloads (mit save_result()) wird dieser Status automatisch auf DONE aktualisiert.

Parallele Workflows​

Für klassische Aufgaben, die parallelisiert werden können, verwende den @distribute_task-Dekorator, um die für die Ausführung einer Aufgabe benötigten Rechenanforderungen zu definieren. Beginne damit, das transpile_remote.py-Beispiel aus dem Thema Dein erstes Qiskit Serverless-Programm schreiben mit dem folgenden Code aufzurufen.

Der folgende Code setzt voraus, dass du deine Zugangsdaten bereits gespeichert hast.

%%writefile ./source_files/transpile_remote.py

from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime import QiskitRuntimeService
from qiskit_serverless import distribute_task

service = QiskitRuntimeService()

@distribute_task(target={"cpu": 1})
def transpile_remote(circuit, optimization_level, backend):
    """Transpiles an abstract circuit (or list of circuits) into an ISA circuit for a given backend."""
    pass_manager = generate_preset_pass_manager(
        optimization_level=optimization_level,
        backend=service.backend(backend)
    )
    isa_circuit = pass_manager.run(circuit)
    return isa_circuit
Writing ./source_files/transpile_remote.py

In diesem Beispiel hast du die Funktion transpile_remote() mit @distribute_task(target={"cpu": 1}) dekoriert. Beim Ausführen wird eine asynchrone parallele Worker-Aufgabe mit einem einzelnen CPU-Kern erstellt und eine Referenz zur Verfolgung des Workers zurückgegeben. Um das Ergebnis abzurufen, übergib die Referenz an die Funktion get(). Damit lassen sich mehrere parallele Aufgaben ausführen:

%%writefile --append ./source_files/transpile_remote.py

from time import time
from qiskit_serverless import get, get_arguments, save_result, update_status, Job

# Get arguments
arguments = get_arguments()
circuit = arguments.get("circuit")
optimization_level = arguments.get("optimization_level")
backend = arguments.get("backend")
Appending to ./source_files/transpile_remote.py
%%writefile --append ./source_files/transpile_remote.py
# Start distributed transpilation

update_status(Job.OPTIMIZING_HARDWARE)

start_time = time()
transpile_worker_references = [
    transpile_remote(circuit, optimization_level, backend)
    for circuit in arguments.get("circuit_list")
]

transpiled_circuits = get(transpile_worker_references)
end_time = time()
Appending to ./source_files/transpile_remote.py
%%writefile --append ./source_files/transpile_remote.py
# Save result, with metadata

result = {
    "circuits": transpiled_circuits,
    "metadata": {
        "resource_usage": {
            "RUNNING: OPTIMIZING_FOR_HARDWARE": {
                "CPU_TIME": end_time - start_time,
                "QPU_TIME": 0,
            },
        }
    },
}

save_result(result)
Appending to ./source_files/transpile_remote.py
# This cell is hidden from users.
# It uploads the serverless program and checks it runs.

def test_serverless_job(title, entrypoint):
    # Import in function to stop them interfering with user-facing code
    from qiskit.circuit.random import random_circuit
    from qiskit_serverless import IBMServerlessClient, QiskitFunction
    import time
    import uuid

    title += "_" + uuid.uuid4().hex[:8]
    serverless = IBMServerlessClient()
    transpile_remote_demo = QiskitFunction(
        title=title,
        entrypoint=entrypoint,
        working_dir="./source_files/",
    )
    serverless.upload(transpile_remote_demo)
    job = serverless.get(title).run(
        circuit=random_circuit(3, 3),
        circuit_list=[random_circuit(3, 3) for _ in range(3)],
        backend="ibm_torino",
        optimization_level=1,
    )
    for retry in range(25):
        time.sleep(5)
        status = job.status()
        if status == "DONE":
            print("Job completed successfully")
            return
        if status not in [
            "QUEUED",
            "INITIALIZING",
            "RUNNING",
            "RUNNING: OPTIMIZING_FOR_HARDWARE",
            "DONE",
        ]:
            raise Exception(
                f"Unexpected job status '{status}'.\nHere's the logs:\n"
                + job.logs()
            )
        print(f"Waiting for job (status '{status}')")
    raise Exception("Job did not complete in time")

test_serverless_job(
    title="transpile_remote_serverless_test", entrypoint="transpile_remote.py"
)
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'RUNNING')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Job completed successfully

Verschiedene Aufgabenkonfigurationen erkunden​

Du kannst CPU, GPU und Arbeitsspeicher für deine Aufgaben flexibel über @distribute_task() zuweisen. Bei Qiskit Serverless auf der IBM Quantum® Platform verfügt jedes Programm über 16 CPU-Kerne und 32 GB RAM, die nach Bedarf dynamisch zugewiesen werden können.

CPU-Kerne können als vollständige oder sogar als anteilige CPU-Kerne zugewiesen werden, wie im Folgenden gezeigt.

Arbeitsspeicher wird in Bytes angegeben. Zur Erinnerung: 1 Kilobyte entspricht 1024 Byte, 1 Megabyte entspricht 1024 Kilobyte und 1 Gigabyte entspricht 1024 Megabyte. Um 2 GB Arbeitsspeicher für deinen Worker zuzuweisen, musst du "mem": 2 * 1024 * 1024 * 1024 angeben.

%%writefile --append ./source_files/transpile_remote.py

@distribute_task(target={
    "cpu": 16,
    "mem": 2 * 1024 * 1024 * 1024
})
def transpile_remote(circuit, optimization_level, backend):
    return None
Appending to ./source_files/transpile_remote.py
# This cell is hidden from users.
# It checks the distributed program works.
test_serverless_job(
    title="transpile_remote_serverless_test", entrypoint="transpile_remote.py"
)
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'RUNNING')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Job completed successfully

Daten programmübergreifend verwalten​

Mit Qiskit Serverless kannst du Dateien im Verzeichnis /data programmübergreifend verwalten. Dabei gelten folgende Einschränkungen:

  • Aktuell werden nur tar- und h5-Dateien unterstützt
  • Es handelt sich um einen flachen /data-Speicher; Unterverzeichnisse wie /data/folder/ sind nicht möglich

Das folgende Beispiel zeigt, wie du Dateien hochlädst. Stelle sicher, dass du dich mit deinem IBM Quantum-Konto bei Qiskit Serverless authentifiziert hast (Anleitung unter Auf der IBM Quantum Platform bereitstellen).

import tarfile
from qiskit_serverless import IBMServerlessClient

# Create a tar
filename = "transpile_demo.tar"
file = tarfile.open(filename, "w")
file.add("./source_files/transpile_remote.py")
file.close()

# Get a reference to a QiskitFunction
serverless = IBMServerlessClient()
transpile_remote_demo = next(
    program
    for program in serverless.list()
    if program.title == "transpile_remote_serverless"
)

# Upload the tar to Serverless data directory
serverless.file_upload(file=filename, function=transpile_remote_demo)
'{"message":"/usr/src/app/media/5e1f442128cdf60018496a04/transpile_demo.tar"}'

Als Nächstes kannst du alle Dateien in deinem data-Verzeichnis auflisten. Diese Daten sind für alle Programme zugänglich.

serverless.files(function=transpile_remote_demo)
['classifier_name.pkl.tar', 'output.json.tar', 'transpile_demo.tar']

Dies kann innerhalb eines Programms mit file_download() erfolgen, um die Datei in die Programmumgebung herunterzuladen und das tar-Archiv zu entpacken.

%%writefile ./source_files/extract_tarfile.py

import tarfile
from qiskit_serverless import IBMServerlessClient

serverless = IBMServerlessClient(token="<YOUR_API_KEY>") # Use the 44-character API_KEY you created and saved from the IBM Quantum Platform Home dashboard
files = serverless.files()
demo_file = files[0]
downloaded_tar = serverless.file_download(demo_file)

with tarfile.open(downloaded_tar, 'r') as tar:
    tar.extractall()

Nun kann dein Programm mit den Dateien interagieren, genau wie bei einem lokalen Experiment. file_upload(), file_download() und file_delete() können sowohl aus deinem lokalen Experiment als auch aus deinem hochgeladenen Programm heraus aufgerufen werden — für eine konsistente und flexible Datenverwaltung.

Nächste Schritte​

Empfehlungen