Pierwsze zadanie workflow

Ten workflow pobiera plik tekstowy lub CSV, liczy jego wiersze i wypisuje wynik w logach joba. Jeśli wskażesz wolumen, skrypt zapisze na nim również plik z wynikiem.

Pełny skrypt

Zobacz pełny skrypt process_file_workflow.py

#!/usr/bin/env python3
"""Download a text/CSV file in a CGC workflow job and count its lines."""

import argparse
import json
import shlex
import time
from pathlib import PurePosixPath

from cgc.sdk import App, Image, Volume


DEFAULT_INPUT_URL = "https://raw.githubusercontent.com/cs109/2014_data/master/countries.csv"


def wait_for_job(run, timeout_seconds: int, poll_seconds: int = 5):
    deadline = time.time() + timeout_seconds
    last_status = None

    while time.time() < deadline:
        status = run.status()
        if status != last_status:
            print("Job status:", status)
            last_status = status
        if status in {"Succeeded", "Failed"}:
            return status
        time.sleep(poll_seconds)

    return "Timeout"


def print_job_logs(run):
    response = run.logs(timestamps=True)
    logs = response.get("details", {}).get("logs", {})

    if not logs:
        print("No logs returned.")
        return

    print("Job logs:")
    for pod_name, containers in logs.items():
        for container_name, text in containers.items():
            print(f"[{pod_name}/{container_name}]")
            print(text.rstrip() or "<empty>")


def create_process_file_task(volume=None):
    app = App(
        name="file-processing-workflow",
        image=Image.from_registry("python:3.11-slim"),
        volumes=[volume] if volume else [],
    )


    @app.function(
        name="count-file-lines",
        cpu=1,
        memory=2,
        ttl_seconds_after_finished=3600,
    )
    def process_file(input_url: str, output_path: str = ""):
        code = f"""
from pathlib import Path
from urllib.request import urlopen

input_url = {json.dumps(input_url)}
output_path = {json.dumps(output_path)}

with urlopen(input_url, timeout=60) as response:
    text = response.read().decode("utf-8")

line_count = len(text.splitlines())
print(f"Downloaded: {{input_url}}")
print(f"Line count: {{line_count}}")

if output_path:
    path = Path(output_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(str(line_count), encoding="utf-8")
    print(f"Saved line count to: {{path}}")
"""
        return f"python -c {shlex.quote(code)}"

    return process_file


def wait_for_volume(volume, timeout_seconds=300):
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        if volume.ready():
            return
        time.sleep(5)
    raise RuntimeError(f"Volume '{volume.name}' did not become ready")


def parse_args():
    parser = argparse.ArgumentParser(
        description="Count lines in a remote text or CSV file using a CGC job."
    )
    parser.add_argument(
        "--input-url",
        default=DEFAULT_INPUT_URL,
        help="Public URL of the text or CSV file processed by the CGC job.",
    )
    parser.add_argument(
        "--volume",
        help="CGC volume name used to persist the result file.",
    )
    parser.add_argument(
        "--volume-size",
        type=int,
        default=20,
        help="Volume capacity in GB when a missing volume is created.",
    )
    parser.add_argument(
        "--storage-class",
        help="Storage class for a new volume; omit to use the configured default.",
    )
    parser.add_argument(
        "--mount-path",
        default="/data",
        help="Path where the volume is mounted inside the job container.",
    )
    parser.add_argument(
        "--output",
        help="Result file path. With a volume, it must be inside --mount-path.",
    )
    parser.add_argument(
        "--delete-job",
        action="store_true",
        help="Delete this job after successful completion and log retrieval.",
    )
    parser.add_argument(
        "--timeout-seconds",
        type=int,
        default=300,
        help="Maximum number of seconds to wait for the job.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.timeout_seconds <= 0:
        raise ValueError("--timeout-seconds must be positive")
    if args.volume:
        mount = PurePosixPath(args.mount_path)
        if not mount.is_absolute() or ".." in mount.parts:
            raise ValueError("--mount-path must be an absolute container path without '..'")
    volume = None
    output_path = args.output or ""

    if args.volume:
        volume = Volume.from_name(
            name=args.volume,
            mount_path=args.mount_path,
            size=args.volume_size,
            storage_class=args.storage_class,
            create_if_missing=True,
        )
        if not output_path:
            output_path = str(PurePosixPath(args.mount_path) / "line-count.txt")
        if ".." in PurePosixPath(output_path).parts or not PurePosixPath(output_path).is_relative_to(
            PurePosixPath(args.mount_path)
        ):
            raise ValueError("--output must be inside --mount-path")
        volume.ensure()
        wait_for_volume(volume, args.timeout_seconds)
    elif output_path:
        print(
            "Warning: no volume was selected. The output file will disappear "
            "when the job container is removed."
        )

    process_file = create_process_file_task(volume)
    run = process_file.submit(args.input_url, output_path)
    print("Created job:", run.name)

    status = wait_for_job(run, timeout_seconds=args.timeout_seconds)

    print("Final status:", status)

    if status == "Timeout":
        print("The job did not finish before the timeout.")
        print("Check platform events with:")
        print(f"cgc events {run.name}")
    else:
        print_job_logs(run)

    if status != "Succeeded":
        raise RuntimeError(f"Job '{run.name}' ended with status {status}; left for diagnosis")

    if args.delete_job:
        run.delete()
        print("Deleted job:", run.name)


if __name__ == "__main__":
    main()

Jeśli job zakończy się błędem (Failed) lub upłynie czas oczekiwania (Timeout), skrypt zgłosi błąd i pozostawi job do sprawdzenia. Opcja --delete-job usuwa job tylko po pomyślnym zakończeniu i odczytaniu logów.

Krok 1: Wybierz obraz

image=Image.from_registry("python:3.11-slim")

python:3.11-slim to publiczny obraz kontenera z Pythonem 3.11. Kod uruchamiany w kontenerze korzysta tylko z biblioteki standardowej Pythona, więc nie wymaga instalowania dodatkowych pakietów.

Krok 2: Utwórz aplikację

app = App(
    name="file-processing-workflow",
    image=...,
    volumes=[volume] if volume else [],
)

App przechowuje ustawienia wspólne dla zadań:

• name to nazwa workflow i prefiks nazw zasobów.
• image to domyślny obraz używany przez zadania workflow.
• volumes wskazuje wolumeny, które zostaną zamontowane w kontenerach zadań.

Samo utworzenie obiektu App nie tworzy zasobów w CGC.

Krok 3: Zdefiniuj zadanie

@app.function(
    name="count-file-lines",
    cpu=1,
    memory=2,
    ttl_seconds_after_finished=3600,
)
def process_file(input_url, output_path=""):
    ...

process_file to przykładowa nazwa funkcji.

Wartość	Znaczenie
name="count-file-lines"	Prefiks nazw tworzonych jobów CGC.
cpu=1	Rezerwuje jeden rdzeń CPU.
memory=2	Rezerwuje 2 GB pamięci.
ttl_seconds_after_finished=3600	Usuwa zakończony job po jednej godzinie.
input_url	Argument Twojej funkcji. Wskazuje plik do przetworzenia.
output_path	Opcjonalna ścieżka pliku z liczbą wierszy. Zapisz go na wolumenie, np. w /data/line-count.txt, aby zachować wynik po usunięciu kontenera.

Krok 4: Przygotuj polecenie dla kontenera

Funkcja przygotowuje kod Pythona z podanym adresem URL i zwraca polecenie, które uruchomi ten kod w kontenerze:

return f"python -c {shlex.quote(code)}"

json.dumps(input_url) bezpiecznie umieszcza URL w kodzie Pythona. shlex.quote(code) zabezpiecza ten kod przed interpretacją przez powłokę.

Sama funkcja uruchamia się lokalnie. Zwrócone polecenie python -c ... uruchamia się w CGC.

Krok 5: Utwórz job

run = process_file.submit("https://example.com/input.csv", "")

submit() przekazuje URL i opcjonalną ścieżkę wyniku do process_file(input_url, output_path), odbiera przygotowane polecenie i tworzy job CGC. Zwraca obiekt JobRun reprezentujący ten konkretny job.

Krok 6: Poczekaj i odczytaj logi

status = wait_for_job(run, timeout_seconds=300)
if status == "Timeout":
    print(f"Check platform events with: cgc events {run.name}")
else:
    print_job_logs(run)

• timeout_seconds=300 ustawia czas oczekiwania na maksymalnie pięć minut.
• logs() zwraca logi z kontenera zadania.
• print_job_logs() odczytuje treść logów z odpowiedzi SDK i wypisuje ją w konsoli.
• timestamps=True dodaje datę i godzinę do wpisów w logach.

Funkcja oczekująca na zakończenie joba zwraca Succeeded, Failed lub Timeout. Upłynięcie czasu oczekiwania nie zatrzymuje joba w CGC. Jeśli job pozostaje w stanie Pending, sprawdź zdarzenia platformy:

cgc events <job-name>

Pending oznacza, że job został przyjęty, ale kontener zadania jeszcze się nie uruchomił. Przyczyną może być pobieranie obrazu, oczekiwanie na dostępne zasoby, przekroczenie limitów, montowanie wolumenu lub błąd uruchamiania.

Utwórz i uruchom skrypt

Utwórz nowy plik we własnym projekcie, na przykład:

process_file_workflow.py

Wklej do niego pełny skrypt z początku tej strony. Możesz wybrać inną nazwę pliku; nie wpływa ona na działanie SDK.

Uruchom plik z katalogu, w którym go utworzono:

python process_file_workflow.py

Ta komenda:

1. Uruchamia skrypt Pythona na Twoim komputerze.
2. Korzysta z adresu pliku CSV podanego w DEFAULT_INPUT_URL.
3. Wywołuje Twoją funkcję workflow process_file.
4. Tworzy job CGC, który pobiera plik i liczy jego wiersze.
5. Czeka na zakończenie joba i wypisuje jego status oraz logi.

Aby zachować wynik po usunięciu kontenera, wskaż wolumen i ścieżkę pliku:

python process_file_workflow.py \
  --volume file-processing-data \
  --storage-class ssd-rwx-ec \
  --output /data/line-count.txt

--storage-class musi wskazywać klasę dostępną w Twoim środowisku; podaj ją, jeśli nie ustawiono klasy domyślnej.

Skrypt korzysta z istniejącego wolumenu lub tworzy nowy i montuje go w kontenerze pod /data. Sprawdza też, czy ścieżka podana w --output prowadzi do pliku na tym wolumenie.

Aby przetworzyć inny publiczny plik tekstowy lub CSV, przekaż --input-url z własnym adresem URL.

Do uruchomienia przykładu potrzebujesz zainstalowanego pakietu cgcsdk, z którego pochodzą te importy:

from cgc.sdk import App, Image, Volume

Pełny skrypt process_file_workflow.py znajdziesz na stronie Przykłady workflow.

W rozdziale Zadania wielokrotnego użytku dowiesz się, jak uruchomić własny skrypt Pythona, podłączyć wolumen, przesłać pliki przez Filebrowser i przetworzyć wiele obrazów.