用于Spark的从Git到GCS的流程：一种适用于代理的Scala SBT构建与上传流水线

任何曾将 Spark 任务从笔记本电脑部署到生产环境的人都知道这一流程。你需要同时处理 Git 命令、SBT 构建、云存储上传以及若干基础设施 API，每一步都需手动操作。其实还有另一种方法：不必通过提示工程向 AI 请求代码片段，而是将整个生命周期交给 AI 智能体，让它自行执行各步骤。借助 Google 的 Agent Development Kit（ADK）、Gemini 2.5 Flash 和 Dataproc Serverless，大型语言模型就能充当 DevOps 工程师和数据工程师的合作伙伴。

第一阶段涵盖了该架构的基础：一个确定性的 Python 流水线，它从 Git 中获取 Scala/Spark 代码库，使用 SBT 在 Java 11 环境下构建大型 JAR 文件，然后将其上传到 Google Cloud Storage。该流水线被设计为可供代理程序作为工具调用的形式。之后，代理层本身会在其基础上添加 ADK 函数工具、带确认机制的提示界面、Dataproc Serverless 交互式会话以及 Streamlit 聊天前端。

实际效果是，原本需要20分钟且容易出错的构建与部署流程，现在变成了一个可靠的命令，代理程序只需根据简单的英文指令即可执行。

Spark 工作中循环处理的痛点

网页开发者可以通过热模块重载获得近乎即时的反馈，而大数据工程师却很少有这样的体验。通常被称为“内部循环”的编辑、构建、部署和测试流程往往会在多个环节出现停滞：

1. 环境不匹配。笔记本电脑上的Java版本与Spark集群中的版本不一致，本地使用的是Java 17，而Dataproc上则是Java 11字节码，这就是典型的例子。
2. 打包过程耗时。使用sbt clean compile assembly命令生成带依赖的胖JAR文件时，可能会占用CPU和内存数分钟之久。
3. 需手动操作云端工具。人们不得不切换到Google Cloud控制台，或输入容易出错的冗长CLI命令。

笔记本配置连接缓慢。在交互式笔记本环境中使用新的JAR文件时，需要深入查找嵌套的Dataproc属性。

这里的思路是：如果整个流程都是以代码形式呈现，且结构清晰，再交给能够理解自然语言指令的代理程序会怎样？以下内容描述了一个自动化的Git到GCS的Scala/SBT构建流程，旨在作为供LLM代理使用的可执行工具。

面向代理安全工具的设计规则

为了让AI代理能够安全地运行构建系统，该工具必须遵循三个原则：

1. 确定性执行。脚本会在开始任何耗时的编译步骤之前先验证环境，例如确认Java 11版本是否可用。
2. 彻底隔离。构建操作在临时目录中进行，完成后这些目录会被删除，从而避免磁盘空间逐渐占用过多。
3. 标准输入与输出。每个阶段都会输出机器可解析的日志，并返回明确的状态信息。

构建与上传过程被分开处理，因此可以分别进行调用、测试和重试。

逐步实现

将构建环境固定为 Java 11

Dataproc Serverless 上的 Spark 运行时需要与 Java 11 具有二进制兼容性。开发者的机器通常默认使用 Java 17 或 21，这种不匹配会导致诸如 UnsupportedClassVersionError: 61.0 vs 55.0 这样的字节码错误。

artifact_builder.py 中的 ArtifactBuilder 类通过运行时检测已安装的 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}")

可靠地上传到云存储

在某些异常的认证情况下，shell层级的 gsutil 或 gcloud 调用可能会静默失败。uploader.py 模块因此改用官方的 Google Cloud Storage Python 客户端库，该库会抛出真正的异常，使得处理流程能够捕获并报告这些问题：

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。

人们无需记住或手动输入这些命令。开发者只需在聊天界面中说明需求，例如“在我的主分支上构建项目并上传到我的 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代理正在Vertex AI上与Gemini 2.5 Flash协同工作。它会读取用户的聊天消息，并自动使用正确的参数调用artifact_pipeline/run_build_and_upload.py。

监控流水线运行情况

云存储检查。上传完成后，生成的较大尺寸JAR文件会存放在目标桶中，随时可以用于Dataproc。在控制台中，编译后的Spark JAR文件会出现在类似gs://<bucket-name>/spark-jobs/spark-serverless-job_test.jar的路径下，大小为186.8 MB。

通过聊天界面进行操作。借助对话式的Streamlit界面，构建与上传流程完全可通过自然语言完成。开发者只需提供Git仓库地址、分支名、存储桶以及目标路径，代理便会自动规划并执行构建工具。当工件构建及GCS上传完成后，代理会根据其操作状态机在启动无服务器笔记本之前请求确认。

优雅地处理故障

用于生产环境的流水线必须能够以易于理解的方式处理故障。当构建过程或环境出现异常时，代理会捕获相关的错误信息，并提供可行的解决方案而非直接崩溃。

Git身份验证问题。若将流水线指向私有仓库，就会遇到身份验证失败的情况，此时会显示Git的标准错误信息（例如Invalid username or token / Password authentication is not supported）。随后会说明如何配置SSH或令牌。

运行时与环境错误。如果在Python环境中出现意外问题，比如缺少依赖项或出现ModuleNotFoundError，该代理会隔离错误、报告精确的堆栈跟踪信息，并提供恢复方法。聊天会话会保持活跃，会有交互式的重试提示，而不会因错误而中断。

值得借鉴的经验

1. 简化流程复杂性。当一个过程包含四个独立步骤时，应将其整合为一个具有明确输入和输出的幂等命令。

为代码调用设计功能。避免在构建脚本中加入交互式提示，让所有参数都能以参数或API数据的形式传递。

为智能体提供稳固的基础。确定性的基础能让大型语言模型可靠地触发复杂构建，而不会生成错误的命令。

做好故障处理规划。完善的错误拦截机制能让智能体诊断认证问题、缺失的文件以及损坏的依赖项，然后向用户解释这些问题。

可靠的智能体系统是从底层逐步构建起来的。用符合Java 11标准的独立Python引擎替代手动编译步骤，就能将缓慢且脆弱的瓶颈转变为可靠的构建模块。一旦Git到GCS的流程被封装成独立且可重复执行的工具，它就能充当人工智能智能体的“手”和“脚”。