Przenoszenie plików z Git do GCS dla Spark: pipeline budowania i przesyłania w Scala przy użyciu SBT, gotowy do pracy z agentami.
Deterministyczny pipeline w Pythonie, który klonuje repozytorium Scala, tworzy gruby plik JAR w języku Java 11 przy użyciu SBT i przesyła go do GCS, zaprojektowany tak, aby agent ADK mógł uruchomić go jako narzędzie.
Każdy, kto przenosił zadańia typu Spark z laptopa do środowiska produkcyjnego, zna tę rutynę. Musisz zarządzać poleceniami Git, procesem budowania w SBT, przesyłaniem danych do chmury oraz kilkoma interfejsami API infrastrukturalnymi, a każdy krok jest wykonywany ręcznie. Istnieje jednak inny sposób: zamiast prosić sztuczną inteligencję o fragmenty kodu (prompt engineering), przekazujesz agentowi AI cały cykl życia aplikacji i pozwalamy mu samodzielnie wykonywać poszczególne kroki. Dzięki Google’s Agent Development Kit (ADK), Gemini 2.5 Flash oraz Dataproc Serverless model językowy może pełnić rolę partnera w dziedzinie DevOps i inżynierii danych.
To pierwsze etap obejmuje podstawy takiego rozwiązania: deterministyczną ścieżkę w Pythonie, która pobiera repozytorium w Scala/Spark z Git, tworzy dużą plik JAR przy użyciu SBT w środowisku Java 11 i wysyła go do Google Cloud Storage. Ścieżka ta jest zaprojektowana tak, aby agent mógł ją wywołać jako narzędzie. Warstwa agenta sama w sobie, z narzędziami funkcyjnymi ADK, interaktywnymi sesjami Dataproc Serverless oraz interfejsem chatowym Streamlit, buduje się później na jej bazie.
Praktycznym rezultatem jest to, że 20-minutowa, podatna na błędy procedura kompilacji i wdrażania staje się niezawodnym poleceniem, które agent może wykonać na podstawie prostej prośby w języku angielskim.
Kłopoty związane z pętlą wewnętrzną w pracy z Sparkiem
Rozwijający strony internetowe otrzymują niemal natychmiastową informację zwrotną dzięki funkcji hot module reloading. Inżynierowie big data rzadko mają do czynienia z czymś podobnym. Cykl edycji, kompilacji, wdrażania i testowania, często nazywany pętlą wewnętrzną, ma tendencję do zatrzymywania się w kilku punktach:
- Niezgodne środowiska. Wersja Javy na laptopie nie jest zgodna z tą na klastrze Spark. Klasycznym przykładem jest użycie Javy 17 lokalnie wobec bajtówki Javy 11 w Dataproc.
- Drogie pakowanie. Tworzenie pliku JAR z zawartością wszystkich zależności za pomocą polecenia
sbt clean compile assemblymoże zużywać znaczną ilość zasobów CPU i pamięci przez kilka minut. - Ręczne operacje w chmurze. Ludzie przechodzą do konsoli Google Cloud lub wpisują długie polecenia w interfejsie linii poleceń, które łatwo jest błędnie zinterpretować.
Idea polega na tym: a co, gdyby cała ta łańcuchowa struktura była kodem, schludnie upakowanym i przekazanym agencie, który przyjmuje instrukcje w zwykłym języku? Poniższe sekcje opisują zautomatyzowany pipeline budowania w stylu Git-to-GCS dla języka Scala/SBT, zaprojektowany jako narzędzie wykonywalne dla agentów opartych na LLM.
Zasady projektowania narzędzi bezpiecznych dla agentów
Aby agent AI mógł bezpiecznie uruchamiać system budowania, narzędzie musi przestrzegać trzech zasad:
- Deterministyczna eksploatacja. Skrypt weryfikuje środowisko, na przykład potwierdzając obecność Javy 11, zanim rozpocznie jakikolwiek kosztowny krok kompilacji.
- Czysta izolacja. Procesy budowania odbywają się w tymczasowych katalogach, które są następnie usuwane, dzięki czemu przestrzeń dyskowa nie jest nadmiernie obciążana.
- Standardowe wejście i wyjście. Każdy etap wypisuje logi możliwe do przetworzenia przez maszynę i zwraca wyraźny status.
Budowa i przesyłanie są traktowane oddzielnie, dzięki czemu każde z nich może być wywołane, przetestowane i spróbowane ponownie niezależnie.
Implementacja, krok po kroku
Przywiązanie budowy do Java 11
Środowiska działania Spark w Dataproc Serverless wymagają kompatybilności binarnej z Java 11. Maszyny programistów często używają domyślnie Java 17 lub 21, a ta niezgodność powoduje błędy bytecode, takie jak UnsupportedClassVersionError: 61.0 vs 55.0.
Klasa ArtifactBuilder w pliku artifact_builder.py unika tego problemu, sprawdzając zainstalowane wersje JDK w czasie wykonywania i przywiązując JAVA_HOME wyłącznie do podprocesu budowy, dzięki czemu domyślne ustawienia systemu pozostają nietknięte:
import os
import subprocess
import logging
class ArtifactBuilder:
"""Builds Scala JAR artifacts from Git repositories using SBT."""
def __init__(self, repo_url: str, branch: str = "main"):
self.repo_url = repo_url
self.branch = branch
def clone_repo(self, dest_dir: str):
"""Clone the given Git repository to the destination directory."""
logging.info(f"Cloning {self.repo_url} (branch={self.branch}) into {dest_dir}")
subprocess.run(["git", "clone", "-b", self.branch, self.repo_url, dest_dir], check=True)
def find_build_dir(self, root_dir: str) -> str:
"""Recursively search for SBT build file."""
for dirpath, _, filenames in os.walk(root_dir):
if "build.sbt" in filenames:
logging.info(f"Detected SBT build file in {dirpath}")
return dirpath
raise FileNotFoundError(f"No build.sbt file found in {root_dir}")
def check_java_installed(self):
"""Ensure Java 11 is available and active in environment."""
env = os.environ.copy()
env["JAVA_HOME"] = "/opt/homebrew/opt/openjdk@11/libexec/openjdk.jdk/Contents/Home"
env["PATH"] = f"/opt/homebrew/opt/openjdk@11/bin:{env.get('PATH', '')}"
result = subprocess.run(
["java", "-version"],
check=True,
env=env,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
output = result.stdout + result.stderr
if "version" in output:
version_str = output.split("version")[1].split()[0].strip('"')
major_version = int(version_str.split(".")[0])
if major_version != 11:
raise EnvironmentError(f"Java 11 required, but detected Java {major_version}")
logging.info("Java 11 verified successfully.")
def build_artifact(self, repo_dir: str) -> str:
"""Compile and assemble fat JAR."""
build_dir = self.find_build_dir(repo_dir)
self.check_java_installed()
env = os.environ.copy()
env["JAVA_HOME"] = "/opt/homebrew/opt/openjdk@11/libexec/openjdk.jdk/Contents/Home"
env["PATH"] = f"/opt/homebrew/opt/openjdk@11/bin:{env.get('PATH', '')}"
logging.info("Running: sbt clean compile assembly...")
subprocess.run(["sbt", "clean", "compile", "assembly"], cwd=build_dir, env=env, check=True)
target_dir = os.path.join(build_dir, "target")
return self._find_file(target_dir, ".jar")
def _find_file(self, directory: str, extension: str) -> str:
for root, _, files in os.walk(directory):
for f in files:
if f.endswith(extension):
return os.path.join(root, f)
raise FileNotFoundError(f"No {extension} file found in {directory}")
Niezawodne publikowanie w chmurze
Wywołania poziomu shella z użyciem gsutil lub gcloud mogą bezgłośnie zawieść w nietypowych sytuacjach autoryzacyjnych. Dlatego moduł uploader.py wykorzystuje oficjalną bibliotekę klienta Pythona do Google Cloud Storage, która generuje rzeczywiste wyjątki, które proces może przechwycić i zgłosić:
from google.cloud import storage
import logging
import os
class ArtifactUploader:
"""Handles secure upload of artifacts to Google Cloud Storage."""
def __init__(self):
self.client = storage.Client()
def upload_to_gcs(self, bucket_name: str, artifact_path: str, dest_path: str):
if not os.path.exists(artifact_path):
raise FileNotFoundError(f"Artifact not found: {artifact_path}")
logging.info(f"Uploading {artifact_path} → gs://{bucket_name}/{dest_path}")
bucket = self.client.bucket(bucket_name)
blob = bucket.blob(dest_path)
blob.upload_from_filename(artifact_path)
logging.info("Upload to GCS completed successfully.")
Łączenie obu w jedną interfejs CLI
W końcu run_build_and_upload.py łączy narzędzie budujące i przesyłające w jeden ArtifactPipeline z deterministycznym interfejsem wiersza poleceń.
Nikt nie musi zapamiętywać ani wpisywać tych poleceń. Programista określa cel w interfejsie czatu, na przykład „Zbuduj mój repozytorium w wersji main i przesuń go do mojego bufora GCS”. Backend agenta wyodrębnia parametry z tego zdania i uruchamia odpowiednie polecenie. Oto co jest wykonywane w tle:
# Executed autonomously in the background by the Agent's backend tool:
python3 artifact_pipeline/run_build_and_upload.py \
--repo "https://github.com/<username>/demo-pipeline.git" \
--branch main \
--bucket "demo-spark-sandbox-bucket" \
--dest "spark-jobs/spark-serverless-job_test.jar" \
--cleanup
Log z terminala backendu z rzeczywistego uruchomienia pokazuje, że agent Google ADK pracuje z Gemini 2.5 Flash w Vertex AI. Odczytuje wiadomość użytkownika i automatycznie wywołuje artifact_pipeline/run_build_and_upload.py z odpowiednimi parametrami.
Obserwowanie działania pipeline’u
Sprawdzenie w Cloud Storage. Gdy proces przesyłania zostanie zakończony, plik JAR trafia do wybranego bucketa, gotowy do podłączenia do Dataproc. W konsoli skompilowany plik JAR Spark pojawia się pod adresem takim jak gs://<bucket-name>/spark-jobs/spark-serverless-job_test.jar o rozmiarze 186,8 MB.
Kierowanie procesem z interfejsu czatowego. Dzięki połączonemu interfejsowi konwersacyjnemu Streamlit, proces budowania i przesyłania odbywa się w całości za pomocą języka naturalnego. Programista dostarcza repozytorium Git, gałąź, bucket oraz ścieżkę docelową, a agent planuje i uruchamia narzędzia do budowania. Gdy budowa artefaktu oraz przesłanie go do GCS zostaną zakończone, agent kieruje się swoją maszyną stanów operacyjnych i prosi o potwierdzenie przed uruchomieniem serwerless notebooka.
Pożądane zachowanie w przypadku błędów
Procesy przeznaczone do użycia w produkcji muszą zawodzić w sposób zrozumiały. Gdy wystąpi błąd podczas budowania lub w środowisku, agent przechwytuje odpowiednie komunikaty błędu i dostarcza praktyczne wskazówki zamiast doprowadzić do awarii.
Problemy z autoryzacją w Git. Jeśli pipeline jest skierowany na prywatny repozytorium lub występuje błąd autoryzacji, zostaje przechwycony standardowy wyjście błędów Git (na przykład Invalid username or token / Password authentication is not supported). Następnie podano instrukcje dotyczące konfiguracji SSH lub tokena.
Błędy w czasie wykonywania i środowisku. Jeśli w środowisku Python wystąpi coś nieoczekiwanego, takiego jak brakująca zależność lub błąd ModuleNotFoundError, agent izoluje ten błąd, podaje dokładny ślad wywołań i proponuje sposoby naprawy. Sesja czatowa pozostaje aktywna, z interaktywnymi prośbami o ponowną próbę, zamiast zostać przerwana wraz z błędem.
Lekcje do zapamiętania
- Zagospodarowanie złożoności domeny. Gdy proces składa się z czterech odrębnych kroków, należy je połączyć w jeden polecenie idempotentne z wyraźnymi danymi wejściowymi i wyjściowymi.
Niezawodne systemy agentowe rozwijają się od dołu do góry. Zastąpienie ręcznych kroków kompilacji izolowanym silnikiem Python zgodnym z Java 11 przekształca powolne i kruche wąskie gardła w niezawodne elementy budulcowe. Gdy pipeline od Git do GCS zostanie zapakowany jako samodzielne, idempotentne narzędzie, może służyć jako „ręce i nogi” agenta AI.
Literatura pokrewna
- Grafy wiedzy dla sztucznej inteligencji agentowej: cztery wzorce wykraczające poza Vector RAG — Wyszukiwanie wektorowe znajduje podobny tekst, a nie powiązane fakty. Cztery wzorce grafowe dla agentów: ugruntowane instrukcje, planowanie, weryfikacja wykonania oraz samokorekta.
- Od prototypu do produkcji: sztuczna inteligencja agentowa z LangChain, LangGraph i Azure — Wyszukiwanie hybrydowe, buforowanie embeddingów, routowanie w LangGraph, mechanizmy ochronne, testy regresji na podstawie zbiorów danych referencyjnych oraz śledzenie działania agenta obsługującego dokumenty podatkowe, wraz z przedstawionymi wynikami.