您如何使用Google DataProc Java Client通过关联的GS存储桶中的jar文件和类来提交Spark作业？

Question

我需要触发Spark Jobs以使用API​​调用从JSON文件聚合数据。我使用spring-boot创建资源。因此，解决方案的步骤如下：

1. 用户发出带有json文件作为输入的POST请求
2. JSON文件存储在与dataproc集群关联的Google存储桶中。
3. 使用指定的jar，类从REST方法内部触发聚集的Spark作业，并且参数为json文件链接。

我希望使用Dataproc的Java Client而不是控制台或命令行来触发作业。你怎么做呢？

Answer 1

我们希望很快能获得有关正式文档的更全面的指南，但要开始使用，请访问以下API概述：https : //developers.google.com/api-client-
library/java/apis/dataproc/v1

它包括到Dataproc javadocs的链接；如果您的html" target="_blank">服务器代表自己的项目而不是最终用户的Google项目进行调用，则您可能希望此处说明的基于密钥文件的服务帐户身份验证创建Credential用于初始化Dataproc客户端存根的对象。

对于特定于dataproc的部分，这仅意味着如果使用Maven，则将以下依赖项添加到Maven pomfile中：

<project>
  <dependencies>
    <dependency>
      <groupId>com.google.apis</groupId>
      <artifactId>google-api-services-dataproc</artifactId>
      <version>v1-rev4-1.21.0</version>
    </dependency>
  </dependencies>
</project>

然后，您将获得如下代码：

Dataproc dataproc = new Dataproc.Builder(new NetHttpTransport(), new JacksonFactory(), credential)
    .setApplicationName("my-webabb/1.0")
    .build();
dataproc.projects().regions().jobs().submit(
    projectId, "global", new SubmitJobRequest()
        .setJob(new Job()
            .setPlacement(new JobPlacement()
                .setClusterName("my-spark-cluster"))
            .setSparkJob(new SparkJob()
                .setMainClass("FooSparkJobMain")
                .setJarFileUris(ImmutableList.of("gs://bucket/path/to/your/spark-job.jar"))
                .setArgs(ImmutableList.of(
                    "arg1", "arg2", "arg3")))))
    .execute();

由于不同的中间服务器可能会进行低级重试，或者您的请求可能会引发IOException，而您不知道提交作业是否成功，因此您可能要执行的另一步骤是生成自己的作业jobId；那么您知道要轮询哪个jobId，以弄清它是否已提交，即使您的请求超时或引发一些未知的异常：

import java.util.UUID;

...

Dataproc dataproc = new Dataproc.Builder(new NetHttpTransport(), new JacksonFactory(), credential)
    .setApplicationName("my-webabb/1.0")
    .build();

String curJobId = "json-agg-job-" + UUID.randomUUID().toString();
Job jobSnapshot = null;
try {
  jobSnapshot = dataproc.projects().regions().jobs().submit(
      projectId, "global", new SubmitJobRequest()
          .setJob(new Job()
              .setReference(new JobReference()
                   .setJobId(curJobId))
              .setPlacement(new JobPlacement()
                  .setClusterName("my-spark-cluster"))
              .setSparkJob(new SparkJob()
                  .setMainClass("FooSparkJobMain")
                  .setJarFileUris(ImmutableList.of("gs://bucket/path/to/your/spark-job.jar"))
                  .setArgs(ImmutableList.of(
                      "arg1", "arg2", "arg3")))))
      .execute();
} catch (IOException ioe) {
  try {
    jobSnapshot = dataproc.projects().regions().jobs().get(
        projectId, "global", curJobId).execute();
    logger.info(ioe, "Despite exception, job was verified submitted");
  } catch (IOException ioe2) {
    // Handle differently; if it's a GoogleJsonResponseException you can inspect the error
    // code, and if it's a 404, then it means the job didn't get submitted; you can add retry
    // logic in that case.
  }
}

// We can poll on dataproc.projects().regions().jobs().get(...) until the job reports being
// completed or failed now.