Git zu GCS für Spark: Ein agentenfähiger Scala-SBT-Build- und Hochlade-Pipeline
Ein deterministischer Python-Pipeline, der ein Scala-Repository kloniert, mit SBT ein Java 11 Fat JAR erstellt und es auf GCS hochlädt, so dass ein ADK-Agent es als Tool ausführen kann.
Jeder, der bereits eine Spark-Anwendung von einem Laptop in die Produktion gebracht hat, kennt den Ablauf. Man muss sich mit Git-Befehlen, einem SBT-Build-Prozess, dem Hochladen in die Cloud-Speicherung sowie mehreren Infrastruktur-APIs herumschlagen, und jeder Schritt erfolgt manuell. Es gibt jedoch eine andere Vorgehensweise: Anstatt einer KI Codeausschnitte anzufordern (Prompt Engineering), überlässt man einem KI-Agenten den gesamten Lebenszyklus und lässt ihn die Schritte selbst ausführen. Mit Google’s Agent Development Kit (ADK), Gemini 2.5 Flash und Dataproc Serverless kann eine LLM als Partner für DevOps- und Datenengineering-Aufgaben dienen.
In dieser ersten Phase werden die Grundlagen dieser Einrichtung behandelt: ein deterministischer Python-Pipeline, der einen Scala/Spark-Repository aus Git herunterlädt, mithilfe von SBT unter Java 11 ein großes JAR erstellt und dieses anschließend in Google Cloud Storage hochlädt. Die Pipeline ist so geschrieben, dass ein Agent sie als Tool aufrufen kann. Auf dieser Basis wird später die eigentliche Agenteschicht mit ADK-Funktionswerkzeugen, einer bestätigungsbedingten Anfrage, interaktiven Dataproc Serverless-Sitzungen sowie einer Streamlit-Chat-Oberfläche aufgebaut.
Das praktische Ergebnis ist, dass eine 20 Minuten dauernde, fehleranfällige Build- und Deployment-Prozedur zu einem zuverlässigen Befehl wird, den ein Agent aufgrund einer einfachen Anfrage in englischer Sprache ausführen kann.
Die Herausforderungen durch den inneren Schleifenprozess in Spark-Arbeiten
Webentwickler erhalten durch das Hot-Module-Reloading nahezu sofortige Rückmeldungen. Big-Data-Ingenieure bekommen so etwas nur selten. Der Bearbeitungs-, Kompilierungs-, Bereitstellungs- und Testzyklus, oft als „innerer Loop“ bezeichnet, gerät häufig an verschiedenen Stellen ins Stocken:
- Nicht übereinstimmende Umgebungen. Die Java-Version auf einem Laptop stimmt nicht mit der auf dem Spark-Cluster überein. Ein klassisches Beispiel ist Java 17 lokal gegenüber Java 11-Bytecode auf Dataproc.
- Kostenintensives Verpacken. Die Erstellung eines „shaded fat JAR“ mit
sbt clean compile assemblykann den CPU- und Speicherverbrauch mehrere Minuten lang in die Höhe treiben. - Manuelle Abläufe in der Cloud. Die Nutzer wechseln zur Google Cloud Console oder geben lange CLI-Befehle ein, die leicht falsch eingegeben werden können.
Die Idee hier: Was wäre, wenn die gesamte Kette aus Code bestünde, ordentlich verpackt und an einen Agenten übergeben würde, der Anweisungen in Alltagssprache versteht? Die folgenden Abschnitte beschreiben ein automatisiertes Git-zu-GCS-Scala/SBT-Build-Pipeline, das als ausführbares Tool für LLM-Agenten konzipiert ist.
Entwurfsregeln für agentensicheres Tooling
Damit ein KI-Agent ein Build-System sicher ausführen kann, muss das Tool drei Prinzipien befolgen:
- Deterministische Ausführung. Das Skript überprüft die Umgebung beispielsweise durch die Bestätigung von Java 11, bevor es mit irgendwelchen aufwändigen Kompilierungsschritten beginnt.
- Klare Isolation. Die Builds finden in temporären Verzeichnissen statt, die danach entfernt werden, sodass sich der Festplattenspeicher nicht ansammelt.
- Standard-Eingabe und -Ausgabe. Jede Phase gibt maschinenlesbare Protokolle aus und gibt einen expliziten Status zurück.
Der Build- und der Upload-Vorgang werden getrennt gehalten, sodass jeder unabhängig aufgerufen, getestet und erneut ausgeführt werden kann.
Implementierung, Schritt für Schritt
Festlegung des Builds auf Java 11
Die Spark-Runtimes in Dataproc Serverless erfordern eine binäre Kompatibilität mit Java 11. Entwicklerrechner verwenden oft standardmäßig Java 17 oder 21, was zu Bytecode-Fehlern wie UnsupportedClassVersionError: 61.0 vs 55.0 führt.
Die Klasse ArtifactBuilder in artifact_builder.py vermeidet dies, indem sie die installierten JDKs zur Laufzeit überprüft und JAVA_HOME ausschließlich für den Build-Subprozess festlegt, sodass die Systemstandardeinstellungen unberührt bleiben:
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}")
Toxisches Veröffentlichen in Cloud Storage
Aufrufe von gsutil oder gcloud auf Shell-Ebene können bei ungewöhnlichen Authentifizierungssituationen stillschweigend fehlschlagen. Daher verwendet das Modul uploader.py stattdessen die offizielle Python-Client-Bibliothek von Google Cloud Storage, die echte Ausnahmen auslöst, die der Pipeline erfasst und melden kann:
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.")
Beides in einer CLI zusammenfassen
Schließlich vereint run_build_and_upload.py den Builder und den Uploader zu einem ArtifactPipeline mit einer deterministischen Befehlszeilenschnittstelle.
Niemand muss sich diese Befehle merken oder eingeben. Der Entwickler gibt seine Absicht im Chat-Frontend an, zum Beispiel „Bilde mein Repository auf main ab und lade es in meinen GCS-Container hoch.“ Der Backend-Teil des Agents extrahiert die Parameter aus diesem Satz und führt den entsprechenden Befehl aus. Folgendes wird im Hintergrund ausgeführt:
# 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
Ein Backend-Terminal-Log aus einem laufenden Betrieb zeigt, dass der Google ADK-Agent mit Gemini 2.5 Flash auf Vertex AI arbeitet. Er liest die Chat-Nachricht des Benutzers und ruft automatisch artifact_pipeline/run_build_and_upload.py mit den richtigen Parametern auf.
Überwachung des Pipeline-Ausführungsverlaufs
Prüfung im Cloud Storage. Sobald das Hochladen abgeschlossen ist, befindet sich die große JAR-Datei im Ziel-Container und kann direkt an Dataproc angehängt werden. In der Konsole erscheint die kompilierte Spark-JAR-Datei unter einem Pfad wie gs://<Bucket-Name>/spark-jobs/spark-serverless-job_test.jar mit einer Größe von 186,8 MB.
Steuerung über eine Chat-Oberfläche. Mit der konversationalen Streamlit-Schnittstelle können der Build und das Hochladen vollständig über natürliche Sprache durchgeführt werden. Der Entwickler gibt das Git-Repository, die Branch, den Bucket sowie den Zielpfad an, und der Agent plant sowie führt die Build-Tools aus. Sobald der Build des Artefakts sowie das Hochladen in GCS abgeschlossen sind, folgt der Agent seiner operativen Zustandsmaschine und bittet um Bestätigung, bevor ein serverloses Notebook gestartet wird.
Sanftes Versagen
Ein für die Produktion vorgesehenes Pipeline muss auf verständliche Weise versagen. Wenn der Build oder die Umgebung fehlschlägt, fängt der Agent die zugrunde liegenden Fehlermeldungen ab und gibt anwendbare Anleitungen anstelle eines Absturzes.
Probleme bei der Git-Authentifizierung. Wenn die Pipeline auf ein privates Repository ausgerichtet ist oder eine Authentifizierungsfehler auftritt, wird Git’s stderr gespeichert (zum Beispiel Invalid username or token / Password authentication is not supported). Anschließend wird erklärt, wie SSH oder ein Token konfiguriert werden können.
Fehler während der Ausführung und im Umfeld. Wenn in der Python-Umgebung etwas Unerwartetes schiefgeht, wie zum Beispiel ein fehlender Abhängigkeit oder ein ModuleNotFoundError, isoliert der Agent den Fehler, gibt die genaue Fehlerausgabe wieder und schlägt Lösungswege vor. Die Chat-Sitzung bleibt aktiv mit interaktiven Anfragen zur erneuten Ausführung, anstatt mit dem Fehler zu enden.
Lektionen für die Zukunft
- Bereinigung der Domänenkomplexität. Wenn ein Prozess aus vier unterschiedlichen Schritten besteht, sollten diese in einen einzigen idempotenten Befehl mit expliziten Eingaben und Ausgaben zusammengefasst werden.
Zuverlässige Agentensysteme entstehen von unten nach oben. Der Ersatz manueller Kompilierungs Schritte durch einen isolierten, Java 11-kompatiblen Python-Engine verwandelt ein langsames, brüchiges Engpass in einen zuverlässigen Baustein. Sobald der Pipeline-Prozess von Git zu GCS als eigenständiges, idempotentes Tool verpackt ist, kann er die „Hände und Füße“ eines KI-Agenten sein.