Пайплайн для переноса проектов из Git в GCS для Spark: сборка и загрузка с использованием Scala, SBT и поддержка работы агентов.
Детерминистичный Python-пайплайн, который копирует репозиторий Scala, собирает тяжелый JAR для Java 11 с использованием SBT и загружает его в GCS, разработанный таким образом, чтобы агент ADK мог запускать его в качестве инструмента.
Каждый, кто переводил проекты на Spark с ноутбука в продакшн, знает этот процесс. Необходимо одновременно работать с командами Git, запускать сборку через SBT, загружать данные в облачное хранилище и использовать различные API инфраструктуры, причем каждый шаг выполняется вручную. Существует более эффективный способ: вместо того чтобы просить ИИ сгенерировать фрагменты кода (метод инжиниринга запросов), вы передаете ИИ-агенту весь цикл жизни проекта и позволяете ему самостоятельно выполнять все шаги. Благодаря набору для разработки агентов от Google (ADK), модели Gemini 2.5 Flash и сервису Dataproc Serverless большая языковая модель может выступать в роли партнера по DevOps и обработке данных.
На этом первом этапе заложены основы данной конфигурации: детерминистическая Python-пайплайн, которая загружает репозиторий на Scala/Spark из Git, создаёт большой JAR-файл с использованием SBT под Java 11 и загружает его в Google Cloud Storage. Пайплайн написан так, чтобы агент мог вызывать его в качестве инструмента. Сам слой агентов, включающий функциональные инструменты ADK, запрос с подтверждением, интерактивные сессии Dataproc Serverless и фронтенд для чата в формате Streamlit, строится на основе этой пайплайн-структуры.
Практический результат заключается в том, что процедура сборки и развертывания, занимающая 20 минут и склонная к ошибкам, превращается в надёжную команду, которую агент может выполнить по простому запросу на английском языке.
Проблемы, связанные с внутренним циклом в работе Spark
Разработчики веб-приложений получают практически мгновенную обратную связь благодаря функции горячей замены модулей. Инженеры, работающие с большими данными, редко сталкиваются с чем-то подобным. Цикл редактирования, сборки, развертывания и тестирования, часто называемый внутренней петлей, склонен застывать на нескольких этапах:
- Различия в средах. Версия Java на ноутбуке не совпадает с версией в кластере Spark. Классическим примером является использование Java 17 локально против байт-кода Java 11 в Dataproc.
- Затратный процесс упаковки. Создание большого JAR-файла с помощью команды
sbt clean compile assemblyможет занять несколько минут и сильно нагрузить процессор и оперативную память. - Ручные действия в облаке. Люди переходят в консоль Google Cloud или вводят длинные команды CLI, которые легко ошибочно интерпретировать.
Идея заключается в следующем: что, если вся цепочка операций будет представлена в виде кода, аккуратно упакованного и переданного агенту, способному понимать инструкции на простом языке? В следующих разделах описывается автоматизированный конвейер сборки из Git в GCS на языках Scala/SBT, разработанный как исполняемый инструмент для агентов на основе больших языков моделирования.
Правила проектирования инструментов, безопасных для агентов
Чтобы ИИ-агент мог безопасно использовать систему сборки, инструмент должен соблюдать три принципа:
- Детерминистичное выполнение. Скрипт сначала проверяет среду, например, подтверждая наличие Java 11, прежде чем начинать дорогостоящие шаги компиляции.
- Чистая изоляция. Сборка происходит в временных каталогах, которые затем удаляются, чтобы дисковое пространство не заполнялось лишними данными.
- Стандартный ввод и вывод. На каждом этапе выводятся логи, понятные для обработки компьютером, и возвращается явный статус.
Процессы сборки и загрузки разделены, чтобы их можно было вызывать, тестировать и перепробовать независимо друг от друга.
Реализация по шагам
Фиксация сборки на Java 11
Среды выполнения Spark в Dataproc Serverless требуют двоичной совместимости с Java 11. На рабочих машинах разработчиков часто установлен Java 17 или 21, и это несоответствие приводит к ошибкам байт-кода, таким как UnsupportedClassVersionError: 61.0 vs 55.0.
Класс ArtifactBuilder в файле artifact_builder.py избегает этой проблемы путем проверки установленных версий JDK во время выполнения и привязки переменной JAVA_HOME только к процессу сборки, тем самым не влияя на системные настройки:
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}")
Надежная публикация в облачное хранилище
Вызовы команд gsutil или gcloud на уровне оболочки могут беззвучно проваливаться в необычных ситуациях аутентификации. Поэтому модуль uploader.py использует официальную библиотеку-клиент Python для Google Cloud Storage, которая генерирует реальные исключения, позволяя конвейеру их перехватывать и отслеживать:
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.")
Объединение обоих процессов в CLI
Наконец, run_build_and_upload.py объединяет инструменты сборки и загрузки в одну структуру ArtifactPipeline с детерминированным интерфейсом командной строки.
Никому не нужно запоминать или вводить эти команды. Разработчик описывает свою цель в интерфейсе чата, например: «Соберите мой репозиторий в ветке main и загрузите его в мой бакет GCS». Бэкенд агента извлекает параметры из этого сообщения и запускает соответствующую команду. Вот что он выполняет на фоне:
# 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
Лог терминала серверной части с процесса в реальном времени показывает, что агент Google ADK работает с Gemini 2.5 Flash в Vertex AI. Он считывает сообщение пользователя и автоматически вызывает artifact_pipeline/run_build_and_upload.py с соответствующими параметрами.
Наблюдение за запуском конвейера
Проверка Cloud Storage. После завершения загрузки большой файл JAR оказывается в целевом бакете и готов к подключению к Dataproc. В консоли скомпилированный файл Spark JAR отображается по пути вроде gs://<bucket-name>/spark-jobs/spark-serverless-job_test.jar объемом 186,8 МБ.
Управление из интерфейса чата. Благодаря встроенному диалоговому интерфейсу Streamlit сборка и загрузка происходят полностью с использованием естественного языка. Разработчик указывает репозиторий Git, ветку, бакет и путь назначения, после чего агент планирует и запускает инструменты сборки. После завершения сборки артефакта и загрузки его в GCS агент согласно своей машине состояний запрашивает подтверждение перед запуском серверлессной нотбуки.
Корректное обработка сбоев
Пайплайн, предназначенный для производственного использования, должен обрабатывать сбои понятным образом. При сбое сборки или окружения агент перехватывает соответствующие сообщения об ошибках и предоставляет практические рекомендации вместо того, чтобы привести к сбою всей системы.
Проблемы аутентификации в Git. Если пайплайн направлен на приватный репозиторий, возникает сбой аутентификации, в результате чего фиксируется сообщение об ошибках из stderr Git (например, Неверный логин или токен / Аутентификация по паролю не поддерживается). Затем указывается, как настроить SSH или использовать токен.
Ошибки во время выполнения и в среде. Если в среде Python происходит нечто непредвиденное, такое как отсутствие зависимости или ошибка ModuleNotFoundError, агент изолирует ошибку, сообщает точный отчет об ошибках и предлагает способы восстановления. Сессия чата остается активной с интерактивными запросами на повторную попытку, вместо того чтобы прерваться из-за сбоя.
Уроки, которые стоит запомнить
- Сокращение сложности домена задач. Когда процесс состоит из четырех отдельных шагов, их следует объединить в одну команду с явными входными и выходными данными.
Надежные агентные системы формируются снизу вверх. Замена ручных шагов компиляции на изолированный движок Python, совместимый с Java 11, превращает медленные и хрупкие препятствия в надежные составляющие. Как только конвейер Git-to-GCS превращается в самодостаточный инструмент с возможностью повторного выполнения, он может служить руками и ногами ИИ-агента.