Zadania wielokrotnego użytku

Gdy uruchomisz już pierwszy job, w ten sam sposób przygotuj własne zadanie:

1. Wybierz obraz zawierający wymagany program.
2. Utwórz App.
3. Dodaj do funkcji dekorator @app.function.
4. Dane, które zmieniają się między uruchomieniami, przekaż jako argumenty funkcji.
5. Zwróć polecenie, które ma zostać wykonane w CGC.
6. Wywołaj submit() z danymi do przetworzenia.

Przetwarzanie obrazów z PyTorch

Pierwszy przykład pokazuje krótki program zapisany bezpośrednio w funkcji workflow. Większy program umieść w osobnym pliku Pythona, a funkcji workflow użyj tylko do uruchomienia go z odpowiednimi argumentami.

1. Utwórz program przetwarzający obraz

Utwórz plik process_image.py we własnym projekcie:

Zobacz pełny skrypt process_image.py

process_image.py

import argparse

import numpy as np
import torch
from PIL import Image


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        required=True,
        help="Path to the source JPEG or PNG inside the job container.",
    )
    parser.add_argument(
        "--output",
        required=True,
        help="Path where the processed image will be saved.",
    )
    args = parser.parse_args()

    image = Image.open(args.input).convert("RGB")
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    pixels = torch.from_numpy(np.array(image)).to(device)
    processed = 255 - pixels

    output = Image.fromarray(processed.cpu().numpy().astype(np.uint8))
    output.save(args.output)

    print(f"Processed image on {device} and saved it to {args.output}")


if __name__ == "__main__":
    main()

Pillow czyta i zapisuje obraz. PyTorch reprezentuje piksele jako tensor i oblicza 255 - pixels, odwracając kolory obrazu. Obliczenia wykonują się na GPU, jeśli CUDA jest dostępna, a w przeciwnym razie na CPU.

W swoim projekcie możesz w ten sposób zmieniać rozmiar obrazów, normalizować je, ładować model PyTorch, wykonywać predykcję oraz rysować lub zapisywać wynik.

Argument	Znaczenie
--input	Ścieżka do obrazu wejściowego JPEG albo PNG w kontenerze joba.
--output	Ścieżka, pod którą zapisany zostanie przetworzony obraz.

2. Uruchom Filebrowser

Filebrowser pozwala przesyłać, pobierać i usuwać pliki przez przeglądarkę. Pliki znajdują się na wolumenie CGC. Uruchom Filebrowser, jeśli jeszcze nie działa:

cgc compute filebrowser create

Skrypt z następnego kroku tworzy wolumen i podłącza go do Filebrowsera. Przy ponownym uruchomieniu korzysta z istniejącego wolumenu i nie montuje go ponownie, jeśli jest już podłączony.

Nie musisz budować obrazu kontenera dla tego przykładu. Wybrany publiczny obraz zawiera już Python, PyTorch, NumPy i CUDA. Workflow instaluje Pillow przed uruchomieniem skryptu. Kod i dane wejściowe są odczytywane z wolumenu.

3. Utwórz skrypt uruchamiający zadania

Utwórz drugi plik o nazwie run_image_workflow.py we własnym projekcie:

Zobacz pełny skrypt run_image_workflow.py

run_image_workflow.py

import argparse
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume, volume as volume_api
from cgc.sdk.exceptions import SDKException


VOLUME_NAME = "image-processing-data"
MOUNT_PATH = "/data"
FILEBROWSER_NAME = "filebrowser"


def is_mount_details(value):
    return (
        isinstance(value, dict)
        and value.get("pvc_name") == VOLUME_NAME
        and value.get("target_template_name") == FILEBROWSER_NAME
        and "mount_path" in value
    )


def get_mount_details_from_error(error):
    for value in error.args:
        if is_mount_details(value):
            return value
        if isinstance(value, dict) and is_mount_details(value.get("details")):
            return value["details"]
    return None


def prepare_volume():
    data = Volume.from_name(
        name=VOLUME_NAME,
        mount_path=MOUNT_PATH,
        size=2,
        storage_class="ssd-rwx-ec",
        create_if_missing=True,
    )
    data.ensure()
    deadline = time.monotonic() + 300
    while not data.ready():
        if time.monotonic() >= deadline:
            raise RuntimeError("Volume did not become ready: " + VOLUME_NAME)
        time.sleep(5)

    try:
        response = volume_api.volume_mount(
            name=VOLUME_NAME,
            target=FILEBROWSER_NAME,
        )
    except SDKException as error:
        details = get_mount_details_from_error(error)
        if details is None and error.reason == "PVC_MOUNT_ALREADY_MOUNTED":
            print("Volume is already mounted to Filebrowser.")
        elif details is not None:
            print(
                "Volume is mounted to Filebrowser. "
                f"Filebrowser mount path: {details['mount_path']}"
            )
        else:
            raise
    else:
        details = response.get("details", {})
        if response["code"] != 200 and not is_mount_details(details):
            raise RuntimeError(f"Could not mount volume to Filebrowser: {response}")
        if "mount_path" in details:
            print("Filebrowser mount path:", details["mount_path"])

    return data


def create_process_image_task(data):
    app = App(
        name="image-processing-workflow",
        image=Image.from_registry("pytorch/pytorch:2.7.1-cuda12.8-cudnn9-runtime"),
        volumes=[data],
    )

    @app.function(
        name="process-image",
        cpu=4,
        memory=16,
        gpu=1,
        gpu_type="nvidia-rtx-a5000",
    )
    def process_image(input_path, output_path):
        return (
            "test -f /data/process_image.py"
            f" && test -f {shlex.quote(input_path)}"
            " && python -m pip install --quiet Pillow"
            " && python /data/process_image.py"
            f" --input {shlex.quote(input_path)}"
            f" --output {shlex.quote(output_path)}"
        )

    return process_image

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--input",
        action="append",
        help="Input image path inside the job container. Repeat for multiple images.",
    )
    parser.add_argument(
        "--output",
        action="append",
        help="Output image path inside the job container. Repeat for multiple images.",
    )
    args = parser.parse_args()
    input_paths = args.input or ["/data/input.jpg"]
    output_paths = args.output or ["/data/output.jpg"]

    if len(input_paths) != len(output_paths):
        parser.error("Provide the same number of --input and --output values.")

    for path in input_paths + output_paths:
        candidate = PurePosixPath(path)
        if ".." in candidate.parts or not candidate.is_relative_to(MOUNT_PATH):
            parser.error("Input and output paths must be inside " + MOUNT_PATH)

    data = prepare_volume()
    process_image = create_process_image_task(data)

    print("Volume is ready and mounted to Filebrowser:", VOLUME_NAME)
    print("Open Filebrowser, open the mounted volume, and upload to its root:")
    print("- process_image.py")
    for input_path in input_paths:
        print(f"- {input_path.removeprefix(MOUNT_PATH + '/')}")
    print()
    print("Use this command to find the Filebrowser URL and app token:")
    print("cgc compute list -d")
    print("Sign in with username 'admin' and the displayed app token.")
    input("Press Enter after all files are uploaded...")

    runs = process_image.for_each(input_paths, output_paths)
    failed = []

    for run in runs:
        status = run.wait(timeout_seconds=600)
        print("Job name:", run.name)
        print("Final status:", status)
        print(run.logs(timestamps=True))

        if status != "Succeeded":
            failed.append(run.name)
        else:
            run.delete()
            print("Deleted successful job:", run.name)

    if failed:
        raise RuntimeError(f"Failed jobs: {', '.join(failed)}")


if __name__ == "__main__":
    main()

Uruchom go:

python run_image_workflow.py

Skrypt:

1. Korzysta z istniejącego wolumenu image-processing-data lub tworzy nowy.
2. Podłącza wolumen do Filebrowsera, jeśli nie jest już podłączony.
3. Prosi o przesłanie process_image.py i obrazów wejściowych.
4. Tworzy osobny job CGC dla każdego obrazu wejściowego.
5. Czeka na zakończenie każdego joba i wypisuje jego status oraz logi.

Workflow montuje ten sam wolumen pod /data, więc przesłane pliki są dostępne w kontenerze joba jako:

/data/process_image.py
/data/input.jpg

Wartość	Znaczenie
VOLUME_NAME	Trwały wolumen używany dla skryptu, obrazu wejściowego i obrazu wynikowego.
MOUNT_PATH="/data"	Ścieżka, pod którą wolumen jest zamontowany w kontenerze joba.
FILEBROWSER_NAME="filebrowser"	Nazwa zasobu, na którym działa Filebrowser.
image=...	Publiczny obraz zawierający Python, PyTorch, NumPy i CUDA.
volumes=[data]	Montuje wolumen ze skryptem i danymi pod /data.
test -f /data/process_image.py	Sprawdza, czy plik ze skryptem istnieje przed jego uruchomieniem.
test -f input_path	Sprawdza, czy wybrany obraz wejściowy istnieje przed uruchomieniem skryptu.
python -m pip install --quiet Pillow	Instaluje bibliotekę używaną do otwierania i zapisywania JPEG albo PNG.
python /data/process_image.py	Uruchamia skrypt przesłany przez Filebrowser.
input_path	Ścieżka obrazu przekazana do process_image.py.
output_path	Ścieżka wyniku przekazana do process_image.py.
cpu=4	Przydziela cztery rdzenie CPU.
memory=16	Przydziela 16 GiB pamięci.
gpu=1	Przydziela jedno GPU.
gpu_type="nvidia-rtx-a5000"	Wybiera wymagany typ GPU.

Po zakończeniu joba /data/output.jpg pozostaje na wolumenie i może zostać pobrany przez Filebrowsera.

Samo zaimportowanie skryptu nie tworzy zasobów. Po odczytaniu logów skrypt usuwa tylko joby zakończone sukcesem. Wolumen i zapisane na nim pliki pozostają.

Gdzie zapisywać pliki?

Pliki w katalogu /data znajdują się na wolumenie image-processing-data. Pozostają dostępne po usunięciu kontenera joba i mogą być używane przez kolejne joby.

Pliki zapisane poza wolumenem znajdują się w kontenerze joba i nie są przechowywane trwale. Na przykład /tmp/output.jpg znika po usunięciu tego kontenera.

W jednym zadaniu workflow własną ścieżkę mount_path można obecnie ustawić tylko przy podłączeniu jednego wolumenu.

Jeśli program nie używa GPU, wybierz obraz bazowy zgodny z CPU i pomiń gpu oraz gpu_type w @app.function. Job użyje wtedy domyślnego gpu=0, ale nadal otrzyma określoną liczbę rdzeni CPU i ilość pamięci.

Buduj własny obraz tylko wtedy, gdy istniejący obraz nie zawiera pakietów, bibliotek systemowych albo narzędzi wymaganych przez program. Dla krótkiego skryptu, którego zależności są już dostępne w wybranym obrazie, wystarczy udostępnić plik ze skryptem na wolumenie.

Zabezpiecz wartości przekazywane do poleceń

Nie wstawiaj ścieżek, adresów URL ani nazw bezpośrednio do poleceń powłoki:

# Unsafe
return f"python /data/process_image.py --input {input_path}"

# Safe
return (
    "python /data/process_image.py"
    f" --input {shlex.quote(input_path)}"
)

Używaj shlex.quote() dla ścieżek i innych wartości wstawianych do poleceń powłoki, aby nie zostały potraktowane jako jej składnia.

Przetwórz wiele obrazów

Prześlij photo-a.jpg i photo-b.jpg na wolumen przez Filebrowsera. Następnie użyj wcześniej zdefiniowanego zadania process_image, aby przetworzyć oba obrazy:

input_paths = [
    "/data/photo-a.jpg",
    "/data/photo-b.jpg",
]
output_paths = [
    "/data/photo-a-processed.jpg",
    "/data/photo-b-processed.jpg",
]

for_each() przekazuje wartości z tych samych pozycji do process_image:

• Pierwszy job otrzymuje /data/photo-a.jpg i /data/photo-a-processed.jpg.
• Drugi job otrzymuje /data/photo-b.jpg i /data/photo-b-processed.jpg.

Powstaną dwa niezależne joby CGC. Każdy uruchomi ten sam skrypt process_image.py z innymi argumentami --input i --output. Przetworzone obrazy zostają na zamontowanym wolumenie i można je pobrać przez Filebrowsera.

Usuń wolumen

Usuń wolumen tylko wtedy, gdy jego skrypty, obrazy wejściowe i wyniki nie są już potrzebne:

data.delete()

warning

Usunięcie wolumenu trwale usuwa wszystkie jego pliki. Usunięcie zakończonego joba nie usuwa wolumenu. Nie usuwaj wolumenu automatycznie na końcu workflow, jeśli chcesz zachować zapisane na nim pliki.

W opisie API znajdziesz wszystkie argumenty konfiguracji joba i metody JobRun.

Pełne skrypty process_image.py i run_image_workflow.py są dostępne na stronie Przykłady workflow.