Gerencie arquivos entre execuções

Um job de workflow gerenciado funciona como um job em lote em um nó de computação HPC. O disco local do job é um espaço temporário. O job pode gravar nele, mas o workflow o exclui quando o job termina. Para manter arquivos depois que o job termina, salve-os no armazenamento de arquivos, que funciona como seu próprio bucket de armazenamento. Os arquivos que você envia e os arquivos que seu workflow salva permanecem no armazenamento de arquivos, e execuções posteriores do workflow podem lê-los.

Cada workflow tem seu próprio armazenamento de arquivos para cada usuário. Outros usuários que executam o mesmo workflow não veem seus arquivos.

Envie um arquivo de entrada

Envie um arquivo do seu notebook local com file_upload() e depois liste os arquivos no armazenamento com 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']

Tipos de arquivo aceitos

Um workflow gerenciado verifica o conteúdo de cada arquivo enviado, não a extensão, e rejeita arquivos de outros tipos:

Aceitos	Rejeitados
Arquivos JSON, de texto simples, YAML e Python; arquivos binários como NumPy (.npy, .npz), QPY e pickle; arquivos tar, gzip e zip	CSV, PNG, PDF, HDF5 e arquivos vazios

Nomes de arquivo

• Use apenas letras, dígitos, sublinhados (_), hifens (-) e pontos (.). Um workflow gerenciado remove do nome outros caracteres, como espaços, quando você baixa ou exclui um arquivo. Por exemplo, você pode enviar my results.json, mas não consegue baixá-lo.

• O armazenamento de arquivos não tem pastas. file_upload() mantém apenas o nome do arquivo, então inputs/params.json é armazenado como params.json.

Leia arquivos em um workflow

Passe o nome do arquivo ao executar o workflow:

run_workflow_remotely.ipynb

job = my_workflow.run(input_file="hamiltonian.npz")

Para habilitar isso, edite o código do seu workflow para ler o arquivo de /function_user_data, que é onde todo job enxerga o armazenamento de arquivos:

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

Dentro de um job, leia e grave arquivos por essa pasta. Os métodos de arquivo do cliente, como file_upload() e file_download(), são para o seu notebook local.

Para ler um arquivo compactado, extraia-o para o disco local:

source_files/my_workflow.py

import tarfile

with tarfile.open(FILE_STORAGE / "inputs.tar.gz") as tar:
    tar.extractall("/tmp/inputs", filter="data")

Grave arquivos temporários em /tmp. O armazenamento de arquivos é um armazenamento de objetos que o job enxerga como uma pasta, então ele é mais lento que o disco local. Faça leituras e gravações pesadas no disco local e copie os arquivos finalizados para o armazenamento de arquivos.

Salve arquivos de saída

Seu notebook local lê os nomes dos arquivos de saída a partir do resultado:

run_workflow_remotely.ipynb

result = job.result()
print(result["output_files"])  # ['energies.npy', 'summary.json']

Para habilitar isso, salve cada arquivo de saída no nível superior de /function_user_data e retorne os nomes dos arquivos com 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"]})

Os arquivos que o job grava fora de /function_user_data, como em /tmp ou na pasta atual, são excluídos quando o job termina.

Baixe os arquivos de saída

Depois que o job termina, baixe cada arquivo com file_download(). Defina target_name para escolher o nome do arquivo local e download_location para escolher uma pasta local existente:

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

Sem target_name, file_download() adiciona um prefixo aleatório ao nome do arquivo local, como downloaded_3f2a9c1b_energies.npy, e retorna o nome que usou.

Baixe várias saídas como um único arquivo compactado

Se um job salva muitos arquivos de saída, ou arquivos em subpastas, agrupe-os em um único arquivo compactado no workflow e baixe apenas esse arquivo:

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

Para habilitar isso, grave as saídas em uma pasta no disco local, adicione a pasta a um arquivo compactado em /tmp e copie o arquivo compactado para o armazenamento de arquivos:

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

Como o arquivo compactado é criado no disco local e copiado em uma única etapa, o armazenamento de arquivos nunca contém um arquivo compactado incompleto. O arquivo compactado mantém a estrutura de pastas e pode conter arquivos de qualquer tipo. Neste exemplo, os arquivos são extraídos para outputs/<run_name>/.

Exclua arquivos

Um workflow gerenciado não exclui arquivos do armazenamento de arquivos, então os arquivos permanecem até que você os exclua com file_delete():

run_workflow_remotely.ipynb

client.file_delete("hamiltonian.npz", my_workflow)

file_delete() informa sucesso mesmo que o arquivo não exista, então confira o nome com files() primeiro. Para excluir todos os arquivos de uma execução, filtre files() pelo prefixo do nome da execução, como descrito na próxima seção:

run_workflow_remotely.ipynb

for name in client.files(my_workflow):
    if name.startswith("theta-0.1_"):
        client.file_delete(name, my_workflow)

Boa prática: dê um nome único aos arquivos de cada execução

Todas as suas execuções de um workflow compartilham o mesmo armazenamento de arquivos, e um arquivo que um job salva substitui qualquer arquivo armazenado com o mesmo nome. Se você iniciar várias execuções ao mesmo tempo e cada uma salvar energies.npy, apenas o último será mantido.

Como boa prática, passe um nome de execução como argumento e inicie o nome de cada arquivo que a execução salva com ele:

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)

Todas as execuções podem ler o mesmo arquivo de entrada, hamiltonian.npz, porque elas apenas o leem. Envie uma entrada compartilhada uma única vez, antes de iniciar as execuções.

Use nomes de execução que você consiga reconhecer depois e que contenham apenas letras, dígitos, sublinhados, hifens e pontos. Com um prefixo para cada execução, você pode encontrar, baixar e excluir os arquivos de uma execução em conjunto.

Boa prática: crie checkpoints em jobs longos

Um job que é cancelado, falha ou atinge o limite de 24 horas perde tudo o que não está no armazenamento de arquivos. Em execuções longas, salve um checkpoint no armazenamento de arquivos em intervalos regulares e retome a partir dele quando executar o workflow novamente com o mesmo nome de execução:

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)

Para continuar uma execução que parou, chame run() novamente com os mesmos argumentos.

Tenha em mente os seguintes pontos:

• Grave o checkpoint primeiro no disco local e depois copie-o. Se o job parar durante uma gravação, o checkpoint armazenado ficará incompleto e a próxima execução não conseguirá lê-lo.

• Escolha um intervalo que corresponda ao custo de uma etapa. Salve com frequência suficiente para não perder muito trabalho, mas não tanta a ponto de a gravação dos checkpoints deixar o job lento.

• Exclua os checkpoints depois que a execução for bem-sucedida. Uma execução posterior com o mesmo nome então começa do início.

Em um job que é cancelado, você também pode salvar resultados parciais em um handler de SIGTERM.

Próximos passos

Recomendações

• Execute uma varredura de parâmetros para iniciar um job para cada entrada.
• Grave logs e trate erros para salvar resultados parciais quando um job é cancelado.