De Git à GCS pour Spark : Un pipeline de construction et de téléchargement en Scala SBT prêt à être utilisé par des agents.
Un pipeline Python déterministe qui clone un dépôt Scala, crée un JAR lourd Java 11 à l’aide de SBT et le télécharge sur GCS, conçu pour que cet agent ADK puisse l’exécuter en tant qu’outil.
Tout ceux qui ont transféré un projet Spark d’un ordinateur portable vers l’environnement de production connaissent bien cette routine. Il faut gérer des commandes Git, une compilation avec SBT, un téléchargement vers un stockage cloud ainsi que plusieurs API d’infrastructure, et chaque étape se fait manuellement. Il existe une autre façon de procéder : au lieu de demander à une intelligence artificielle des extraits de code (ingénierie de prompts), on confie tout le cycle de vie à un agent IA qui exécute les étapes lui-même. Grâce au Google Agent Development Kit (ADK), à Gemini 2.5 Flash et à Dataproc Serverless, un LLM peut agir en tant que partenaire DevOps et d’ingénierie des données.
Cette première étape couvre les bases de cette configuration : un pipeline Python déterministe qui récupère un dépôt Scala/Spark depuis Git, crée un JAR volumineux à l’aide de SBT sous Java 11, et le met en stockage sur Google Cloud Storage. Le pipeline est conçu de manière à ce qu’un agent puisse l’appeler en tant qu’outil. La couche d’agent elle-même, avec les outils fonctionnels ADK, une demande conditionnée par une confirmation, des sessions interactives Dataproc Serverless et une interface de chat Streamlit, s’installe par la suite sur cette base.
Le résultat pratique est que une routine de construction et de déploiement longue de 20 minutes et sujette aux erreurs devient une commande fiable que l’agent peut exécuter à partir d’une demande formulée en langue anglaise simple.
Les difficultés liées aux boucles internes dans le travail avec Spark
Les développeurs web reçoivent un retour quasi instantané grâce au hot module reloading. Les ingénieurs en big data ont rarement accès à un tel avantage. Le cycle d’édition, de compilation, de déploiement et de test, souvent appelé boucle interne, a tendance à stagner à plusieurs étapes :
- Environnements incompatibles. La version de Java sur un ordinateur portable est incompatible avec celle du cluster Spark. L’exemple classique est Java 17 localement contre du bytecode Java 11 sur Dataproc.
- Emballage coûteux en ressources. La création d’un JAR volumineux avec
sbt clean compile assemblypeut consommer beaucoup de CPU et de mémoire pendant plusieurs minutes. - Manipulations manuelles dans le cloud. Les utilisateurs doivent se rendre sur la Google Cloud Console ou taper de longues commandes CLI qui sont facilement erronées.
L’idée ici : que se passerait-il si toute la chaîne était constituée de code, soigneusement structuré, et transmis à un agent capable d’interpréter des instructions en langage courant ? Les sections suivantes décrivent un pipeline de construction automatisé Git-to-GCS en Scala/SBT conçu comme outil exécutable pour des agents LLM.
Règles de conception pour des outils sûrs pour les agents
Afin qu’un agent IA puisse exécuter un système de construction en toute sécurité, l’outil doit respecter trois principes :
- Exécution déterministe. Le script valide l’environnement, par exemple en s’assurant que Java 11 est présent, avant de commencer toute étape de compilation coûteuse.
- Isolement propre. Les constructions ont lieu dans des dossiers temporaires qui sont supprimés par la suite, afin que l’espace disque ne s’accumule pas.
- Entrée et sortie standard. Chaque étape affiche des journaux pouvant être analysés par la machine et renvoie un statut explicite.
La construction et le téléversement sont séparés afin que chacun puisse être exécuté, testé et réessayé indépendamment.
Mise en œuvre, étape par étape
Fixation de la construction sur Java 11
Les environnements d’exécution Spark sur Dataproc Serverless nécessitent une compatibilité binaire avec Java 11. Les ordinateurs des développeurs utilisent souvent par défaut Java 17 ou 21, ce qui provoque des erreurs de bytecode telles que UnsupportedClassVersionError: 61.0 vs 55.0.
La classe ArtifactBuilder dans artifact_builder.py évite ce problème en inspectant les JDK installés au moment de l’exécution et en liant JAVA_HOME uniquement au sous-processus de construction, de sorte que la configuration par défaut du système reste inchangée :
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}")
Publier de manière fiable dans le stockage cloud
Les appels au niveau de la shell avec gsutil ou gcloud peuvent échouer silencieusement dans certaines situations d’authentification inhabituelles. Le module uploader.py utilise donc la bibliothèque client officielle Python pour Google Cloud Storage, qui génère des exceptions réelles que le pipeline peut intercepter et signaler :
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.")
Intégration des deux dans une interface en ligne de commande
Finalement, run_build_and_upload.py combine le générateur et l’uploadeur en un seul ArtifactPipeline doté d’une interface en ligne de commande déterministe.
Nul n’a besoin de mémoriser ou de taper ces commandes. Le développeur exprime son intention dans l’interface de chat, par exemple "Générer mon dépôt sur la branche main et l’uploader dans mon bucket GCS." Le backend de l’agent extrait les paramètres de cette phrase et exécute la commande correspondante. Voici ce qui est mis en œuvre en arrière-plan :
# 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
Un journal de terminal backend provenant d’une exécution en temps réel montre l’agent Google ADK fonctionnant avec Gemini 2.5 Flash sur Vertex AI. Il lit le message de chat de l’utilisateur et appelle automatiquement artifact_pipeline/run_build_and_upload.py avec les paramètres appropriés.
Surveillance de l’exécution du pipeline
Vérification dans Cloud Storage. Une fois l’opération d’upload terminée, le JAR volumineux se trouve dans le bucket cible, prêt à être associé à Dataproc. Dans la console, le JAR Spark compilé apparaît sous un chemin du type gs://<bucket-name>/spark-jobs/spark-serverless-job_test.jar, d’une taille de 186,8 MB.
L’utilisation depuis une interface de chat. Grâce à l’interface conversationnelle Streamlit, la compilation et le téléchargement s’effectuent entièrement à partir du langage naturel. Le développeur fournit le répertoire Git, la branche, le bucket ainsi que le chemin de destination, et l’agent planifie et exécute les outils de compilation. Lorsque la compilation de l’artefact et le téléchargement sur GCS sont terminés, l’agent suit sa machine à états opérationnelle et demande une confirmation avant de lancer un notebook serverless.
Faillir de manière contrôlée
Un pipeline destiné à la production doit pouvoir échouer de manière lisible. Lorsque la compilation ou l’environnement tombe en panne, l’agent intercepte les erreurs stderr sous-jacentes et fournit des indications concrètes au lieu de planter le système.
Problèmes d’authentification Git. Si le pipeline est orienté vers un répertoire privé, une erreur d’authentification se produit, ce qui entraîne l’enregistrement des erreurs stderr de Git (par exemple Invalid username or token / Password authentication is not supported). Il explique ensuite comment configurer SSH ou un jeton d’authentification.
Erreurs en temps de exécution et d’environnement. En cas de problème inattendu dans l’environnement Python, comme l’absence d’une dépendance ou une ModuleNotFoundError, l’agent isole l’erreur, affiche le rapport d’erreur complet et propose des moyens de la résoudre. La session de chat reste active, avec des invites interactives pour réessayer, au lieu de s’arrêter avec l’échec.
Leçons à retenir
- Gérer la complexité des domaines. Lorsqu’un processus comporte quatre étapes distinctes, il convient de les regrouper en une seule commande idempotente avec des entrées et des sorties explicites.
Les systèmes d’agents fiables se développent de bas en haut. Remplacer les étapes de compilation manuelles par un moteur Python isolé, compatible Java 11, transforme un goulot d’étranglement lent et fragile en un élément de construction fiable. Une fois le pipeline Git-to-GCS emballé en outil autonome et idempotent, il peut servir de bras et de jambes à un agent IA.