De Git a GCS para Spark: Un pipeline de compilación y subida en Scala con SBT listo para agentes.
Un pipeline determinista en Python que clona un repositorio de Scala, genera un JAR grande para Java 11 con SBT y lo sube a GCS, diseñado para que un agente ADK pueda ejecutarlo como herramienta.
Cualquiera que haya llevado un proyecto de Spark desde una laptop a producción conoce bien el proceso. Hay que manejar comandos de Git, procesos de compilación con SBT, subidas a almacenamiento en la nube y varias APIs de infraestructura, y cada paso se realiza manualmente. Existe una forma diferente de hacerlo: en lugar de pedirle a una IA fragmentos de código (ingeniería de prompts), se le entrega todo el ciclo de vida a un agente de IA para que ejecute los pasos por sí mismo. Con el Kit de Desarrollo de Agentes de Google (ADK), Gemini 2.5 Flash y Dataproc Serverless, un LLM puede funcionar como socio en tareas de DevOps e ingeniería de datos.
Esta primera etapa abarca los fundamentos de esa configuración: un pipeline determinista en Python que obtiene un repositorio en Scala/Spark desde Git, genera un JAR de gran tamaño con SBT bajo Java 11 y lo sube a Google Cloud Storage. El pipeline está diseñado de tal manera que un agente pueda llamarlo como herramienta. La capa del agente en sí, con las herramientas funcionales de ADK, un prompt controlado por confirmación, sesiones interactivas de Dataproc Serverless y una interfaz de chat en Streamlit, se construye posteriormente sobre él.
El resultado práctico es que una rutina de compilación e implementación de 20 minutos, propensa a errores, se convierte en una orden fiable que un agente puede ejecutar a partir de una solicitud en inglés sencillo.
Las dificultades del bucle interno en el trabajo con Spark
Los desarrolladores web reciben retroalimentación casi instantánea gracias al recargue dinámico de módulos. Los ingenieros de big data rara vez obtienen algo similar. El ciclo de edición, compilación, despliegue y pruebas, a menudo llamado bucle interno, tiende a detenerse en varios puntos:
- Entornos incompatibles. La versión de Java en una laptop no coincide con la del clúster Spark. Un ejemplo clásico es usar Java 17 localmente frente al bytecode de Java 11 en Dataproc.
- Procesos de empaquetado costosos. Generar un JAR grande con
sbt clean compile assemblypuede consumir gran cantidad de CPU y memoria durante varios minutos. - Procedimientos manuales en la nube. La gente recurre a la Google Cloud Console o escribe comandos de línea de comando largos que son fáciles de errores.
La idea aquí es: ¿qué pasaría si toda la cadena estuviera codificada de manera ordenada y se le entregara a un agente que acepte instrucciones en lenguaje sencillo? Las secciones siguientes describen un pipeline automatizado de compilación de Git a GCS en Scala/SBT, diseñado como una herramienta ejecutable para agentes de LLM.
Reglas de diseño para herramientas seguras para agentes
Para que un agente de IA ejecute un sistema de compilación de forma segura, la herramienta debe seguir tres principios:
- Ejecución determinista. El script valida el entorno, por ejemplo, confirmando que se utilice Java 11, antes de iniciar cualquier paso de compilación costoso.
- Aislamiento limpio. Las compilaciones se realizan en directorios temporales que se eliminan posteriormente, para que el espacio en disco no se acumule.
- Entrada y salida estándar. Cada etapa imprime registros que pueden ser analizados por la máquina y devuelve un estado explícito.
La compilación y la carga se mantienen separadas para que cada una pueda ser llamada, probada e intentada nuevamente de forma independiente.
Implementación, paso a paso
Fijar la compilación en Java 11
Los entornos de ejecución de Spark en Dataproc Serverless requieren compatibilidad binaria con Java 11. Las máquinas de los desarrolladores suelen usar por defecto Java 17 o 21, y esa discrepancia genera errores de bytecode como UnsupportedClassVersionError: 61.0 vs 55.0.
La clase ArtifactBuilder en artifact_builder.py evita esto al inspeccionar los JDK instalados en tiempo de ejecución y vincular JAVA_HOME únicamente al subproceso de compilación, de modo que el valor por defecto del sistema permanece sin cambios:
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}")
Publicación fiable en almacenamiento en la nube
Las llamadas a nivel de shell con gsutil o gcloud pueden fallar silenciosamente en situaciones de autenticación inusuales. Por lo tanto, el módulo uploader.py utiliza en su lugar la biblioteca de cliente oficial para Python de Google Cloud Storage, la cual genera excepciones reales que el proceso puede capturar y reportar:
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.")
Integrar ambos en una CLI
Finalmente, run_build_and_upload.py combina el generador y el cargador en una sola ArtifactPipeline con una interfaz de línea de comandos determinista.
Nadie necesita memorizar ni escribir estos comandos. El desarrollador expresa su intención en la interfaz de chat, por ejemplo “Generar mi repositorio en la rama main y cargarlo en mi bucket de GCS”. El backend del agente extrae los parámetros de esa frase y ejecuta el comando correspondiente. Esto es lo que se procesa en segundo plano:
# 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 registro de la terminal de backend de una ejecución en tiempo real muestra que el agente Google ADK funciona con Gemini 2.5 Flash en Vertex AI. Este lee el mensaje de chat del usuario y llama automáticamente a artifact_pipeline/run_build_and_upload.py con los parámetros adecuados.
Observando la ejecución del pipeline
Verificación en Cloud Storage. Una vez que finaliza la carga, el archivo JAR se encuentra en el bucket de destino, listo para ser adjuntado a Dataproc. En la consola, el JAR compilado de Spark aparece en una ruta como gs://<bucket-name>/spark-jobs/spark-serverless-job_test.jar, con un tamaño de 186.8 MB.
Uso desde una interfaz de chat. Con la interfaz conversacional de Streamlit integrada, la compilación y la carga se realizan íntegramente mediante lenguaje natural. El desarrollador proporciona el repositorio Git, la rama, el bucket y la ruta de destino, y el agente planifica y ejecuta las herramientas de compilación. Una vez completada la generación del artefacto y la carga en GCS, el agente sigue su máquina de estados operativa y solicita confirmación antes de iniciar una notebook sin servidor.
Fallo elegante
Un pipeline destinado a producción debe fallar de manera comprensible. Cuando la compilación o el entorno se ven afectados, el agente intercepta los mensajes de error y ofrece indicaciones útiles en lugar de colapsar.
Problemas de autenticación de Git. Si se dirige el pipeline a un repositorio privado, se produce un fallo de autenticación, lo que genera los mensajes de error de Git (por ejemplo, Invalid username or token / Password authentication is not supported). Luego se explica cómo configurar SSH o un token.
Errores en tiempo de ejecución y del entorno. Si ocurre algo inesperado en el entorno de Python, como la falta de una dependencia o un ModuleNotFoundError, el agente aísla el error, muestra el rastro completo y sugiere formas de recuperarse. La sesión de chat permanece activa, con solicitudes interactivas para intentar nuevamente, en lugar de terminar con el fallo.
Lecciones a llevar adelante
- Reducir la complejidad del dominio. Cuando un proceso consta de cuatro pasos distintos, agrúpelos en una sola orden idempotente con entradas y salidas explícitas.
Los sistemas de agentes fiables se desarrollan de abajo hacia arriba. Sustituir los pasos manuales de compilación por un motor de Python aislado y compatible con Java 11 convierte un cuello de botella lento y frágil en un componente confiable. Una vez que el pipeline de Git a GCS se empaqueta como una herramienta autónoma e idempotente, puede funcionar como las manos y los pies de un agente de IA.
Lecturas relacionadas
- Gráficos de conocimiento para la IA agencial: cuatro patrones que van más allá del RAG vectorial — La búsqueda vectorial encuentra texto similar, no hechos conectados. Cuatro patrones de gráficos para agentes: estimulación basada en datos reales, planificación, validación de ejecución y autocorrección.
- De prototipo a producción: IA agencial con LangChain, LangGraph y Azure — Búsqueda híbrida, cachés de embeddings, enrutamiento con LangGraph, interruptores de circuito, pruebas de regresión con conjuntos de datos de referencia y seguimiento para un agente que maneja documentos fiscales, con resultados presentados.