Пайплайн для перенесення проектів з Git до GCS у Spark: система створення та завантаження проектів на Scala з використанням SBT, готова до роботи з агентами.

Кожен, хто виконував завдання з використанням 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

Розробники веб-сайтів отримують майже миттєву зворотну відповідь завдяки функції швидкого перезавантаження модулів. Інженери з обробки великих даних рідко мають щось подібне. Цикл редагування, збирання, розгортання та тестування, який часто називають внутрішньою петлею, схильний зупинятися в кількох моментах:

1. Несумісні середовища. Версія Java на ноутбуці не сумісна з версією на кластері Spark. Класичним прикладом є використання Java 17 локально проти байткоду Java 11 у Dataproc.
2. Тривала обробка пакетів. Створення великого JAR-файлу за допомогою команди sbt clean compile assembly може споживати значні ресурси CPU та пам’яті протягом кількох хвилин.
3. Ручна робота в хмарі. Люди переходять до консолі Google Cloud або вводять довгі команди CLI, які легко помилитися.

Повільне підключення ноутбука. Підключення нового JAR до середовища інтерактивного ноутбука означає необхідність пошуку серед вкладених властивостей Dataproc.

Ідея полягає у наступному: а що, якби вся ця ланцюгова структура була кодом, акуратно упакованим та переданим агенту, який розуміє інструкції простою мовою? У наступних розділах описується автоматизований конвеєр збірки Git-to-GCS для Scala/SBT, створений як виконуваний інструмент для агентів LLM.

Правила проектування інструментів, безпечних для агентів

Щоб AI-агент міг безпечно використовувати систему збірки, інструмент має дотримуватися трьох принципів:

1. Детерміністична реалізація. Скрипт перевіряє середовище, наприклад, підтверджуюча наявність Java 11, перш ніж розпочати будь-який трудомісткий крок компіляції.
2. Чиста ізоляція. Збірка відбувається у тимчасових каталогах, які після цього видаляються, щоб простір на диску не накопичувався.
3. Стандартний вхід та вихід. Кожен етап виводить логи, зрозумілі для обробки машиною, та повертає чіткий статус.

Процеси будування та завантаження тримаються окремо, щоб кожен з них можна було викликати, тестувати та перепробувати незалежно.

Реалізація, крок за кроком

Фіксація версії 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. У консолі скомпільований JAR Spark з’являється за шляхом на кшталт gs://<bucket-name>/spark-jobs/spark-serverless-job_test.jar розміром 186,8 MB.

Керування з інтерфейсу чату. Завдяки інтерфейсу Streamlit у форматі розмови всі операції зі створення та завантаження виконуються повністю за допомогою природної мови. Розробник надає репозиторій Git, гілку, бакет та шлях призначення, а агент планує та запускає інструменти для створення продукту. Коли процес створення артефакта та завантаження у GCS завершується, агент діє згідно зі своєю машиною станів та запитує підтвердження перед запуском безсерверного ноутбука.

Граційне оброблення помилок

Пайплайн, призначений для продакшену, має обробляти помилки у зрозумілий спосіб. Коли виникають проблеми під час створення продукту або у середовищі, агент перехоплює відповідні повідомлення про помилки та надає корисні рекомендації замість того, щоб програма завершувалася.

Проблеми з автентифікацією в Git. Якщо конвеєр налаштований на приватний репозиторій або виникає помилка автентифікації, це призводить до запису помилок у stderr Git (наприклад Невірне ім’я користувача або токен / Аутентифікація за паролем не підтримується). Далі пояснюється, як налаштувати SSH або токен.

Помилки під час виконання та у середовищі. Якщо в середовищі Python трапляється щось несподіване, наприклад відсутня залежність або помилка ModuleNotFoundError, агент ізолює цю помилку, надає точний звіт про її походження та пропонує способи відновлення. Сесія чату залишається активною з інтерактивними запитами на повторну спробу, замість того щоб закінчитися разом із помилкою.

Уроки, які варто запам’ятати

1. Зменшення складності домену. Коли процес складається з чотирьох окремих кроків, їх слід об’єднати в одну команду, яка є ідемпотентною та має чіткі вхідні та вихідні дані.

Дизайн для викликів з коду. Уникайте інтерактивних запитів у скриптах збірки та робіть кожен параметр доступним як аргумент чи дані API.

Надайте агентам міцну основу. Детерміністична база дозволяє ШІ надійно запускати складну збірку, замість того щоб генерувати хибні команди.

Плануйте на випадок збоїв. Ефективне перехоплення помилок дозволяє агенту діагностувати проблеми автентифікації, відсутні файли та пошкоджені залежності, а потім пояснити їх користувачеві.

Надійні агентські системи формуються знизу догори. Заміна ручних кроків компіляції на ізольований двигун Python, сумісний з Java 11, перетворює повільну та крихку проблему на надійний елемент. Коли конвеєр Git-to-GCS стає самостійним та ідемпотентним інструментом, він може слугувати руками та ногами AI-агента.