如何用Scala通过REST API调用Google Cloud ML Engine的getConfig方法

刚好之前做过类似的需求，给你整理了一个完整的Scala实现示例，包括GCP认证和API请求的核心逻辑。Google Cloud ML Engine的getConfig接口需要先通过服务账号获取合法的访问令牌，然后才能发起REST请求，咱们一步步来：

1. 准备依赖 #

首先需要在你的build.sbt里添加必要的依赖，包括HTTP客户端和GCP认证库：

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-http" % "10.2.10",
  "com.typesafe.akka" %% "akka-stream" % "2.6.20",
  "com.google.auth" % "google-auth-library-oauth2-http" % "1.18.0",
  "com.google.code.gson" % "gson" % "2.10.1" // 用来解析格式化JSON响应
)

2. 完整代码实现 #

下面是包含认证和请求的完整代码，你只需要替换其中的项目ID、模型名称、版本名称以及服务账号密钥文件的路径：

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, HttpResponse, Uri}
import akka.http.scaladsl.unmarshalling.Unmarshal
import akka.stream.ActorMaterializer
import com.google.auth.oauth2.ServiceAccountCredentials
import scala.concurrent.{Await, ExecutionContextExecutor}
import scala.concurrent.duration._
import com.google.gson.Gson

object MlEngineConfigFetcher {
  def main(args: Array[String]): Unit = {
    // 替换为你的实际参数
    val projectId = "your-gcp-project-id"
    val modelName = "your-model-name"
    val versionName = "your-version-name" // 留空则获取模型级别的配置
    val serviceAccountKeyPath = "/path/to/your/service-account-key.json"

    // 初始化Akka运行环境
    implicit val system: ActorSystem = ActorSystem("MlEngineConfigFetcher")
    implicit val materializer: ActorMaterializer = ActorMaterializer()
    implicit val executionContext: ExecutionContextExecutor = system.dispatcher

    try {
      // 1. 获取GCP访问令牌
      val credentials = ServiceAccountCredentials.fromStream(
        new java.io.FileInputStream(serviceAccountKeyPath)
      )
      // 指定ML Engine API的权限范围
      val scopes = List("https://www.googleapis.com/auth/cloud-platform")
      credentials.refreshIfExpired()
      val accessToken = credentials.getAccessToken.getTokenValue

      // 2. 构建getConfig的请求URL
      val baseUri = Uri(s"https://ml.googleapis.com/v1/projects/$projectId/models/$modelName")
      val requestUri = if (versionName.nonEmpty) {
        baseUri.withPath(baseUri.path / "versions" / versionName / "getConfig")
      } else {
        baseUri.withPath(baseUri.path / "getConfig")
      }

      // 3. 发送GET请求
      val request = HttpRequest(uri = requestUri)
        .addHeader("Authorization", s"Bearer $accessToken")

      val responseFuture = Http().singleRequest(request)

      // 4. 处理响应结果
      val response = Await.result(responseFuture, 10.seconds)
      if (response.status.isSuccess) {
        val responseBody = Await.result(Unmarshal(response).to[String], 5.seconds)
        // 格式化输出JSON响应
        val gson = new Gson().newBuilder().setPrettyPrinting().create()
        val formattedConfig = gson.fromJson(responseBody, classOf[Object])
        println("获取到的ML Engine配置：")
        println(gson.toJson(formattedConfig))
      } else {
        println(s"请求失败，状态码：${response.status}")
        val errorBody = Await.result(Unmarshal(response).to[String], 5.seconds)
        println("错误详情：" + errorBody)
      }
    } catch {
      case e: Exception =>
        println("执行过程中发生异常：" + e.getMessage)
        e.printStackTrace()
    } finally {
      // 关闭Akka系统释放资源
      Await.result(system.terminate(), 5.seconds)
    }
  }
}

3. 关键注意事项 #

• 权限配置：你的服务账号需要拥有ml.models.get或ml.versions.get权限，建议直接给服务账号添加Cloud ML Engine Developer角色，或者更细粒度的自定义权限。
• 配置级别选择：如果要获取整个模型的通用配置，直接留空versionName即可；如果需要特定版本的配置，填写对应版本名称即可。
• 令牌自动刷新：代码中调用了refreshIfExpired()，会自动处理令牌过期的情况，无需手动维护令牌生命周期。
• 异步优化：示例用Await做了同步简化处理，生产环境建议改用异步回调或Future链式调用，避免阻塞线程。

内容的提问来源于stack exchange，提问作者ryekos