GCP云函数提交DataProc Spark任务的凭证配置与JobId设置问题

凭证相关说明 #

1. GCP云函数运行时会自动加载应用默认凭证（ADC），不需要你手动构造或传入凭证文件。客户端会自动读取环境内置的云函数服务账号凭证，无需额外配置。
2. 该功能和云函数与DataProc是否同属一个网络没有关联，凭证属于IAM权限管控范畴，和网络连通性是两个独立逻辑。
3. 你只需要给云函数的默认运行服务账号 {项目ID}@appspot.gserviceaccount.com 授予 roles/dataproc.job.submitter 权限，即可正常提交作业。

新版Dataproc客户端设置自定义JobId的方法 #

你使用的google-cloud-dataproc依赖支持通过JobReference设置自定义JobId，只需要在构造Job对象时传入配置好的JobReference即可，修改后的代码片段如下：

// 构造自定义JobId的JobReference
JobReference jobRef = JobReference.newBuilder()
    .setProjectId("你的GCP项目ID")
    .setJobId("spark-job-" + UUID.randomUUID().toString()) // 此处传入你自定义的JobId
    .build();

// 构造Job对象时关联JobReference
Job job = Job.newBuilder()
    .setReference(jobRef) // 新增这一行即可完成自定义JobId配置
    .setPlacement(jobPlacement)
    .setSparkJob(sparkJob)
    .build();

适配HTTP触发云函数的完整实现 #

依赖补充 #

除了你已有的Dataproc依赖，还需要引入云函数Java框架依赖：

<dependency>
    <groupId>com.google.cloud.functions</groupId>
    <artifactId>functions-framework-api</artifactId>
    <version>1.0.4</version>
    <scope>provided</scope>
</dependency>

函数代码示例 #

import com.google.cloud.functions.HttpFunction;
import com.google.cloud.functions.HttpRequest;
import com.google.cloud.functions.HttpResponse;
import com.google.cloud.dataproc.v1.Job;
import com.google.cloud.dataproc.v1.JobControllerClient;
import com.google.cloud.dataproc.v1.JobControllerSettings;
import com.google.cloud.dataproc.v1.JobPlacement;
import com.google.cloud.dataproc.v1.JobReference;
import com.google.cloud.dataproc.v1.SparkJob;
import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ExecutionException;

public class SubmitSparkJobFunction implements HttpFunction {
    // 替换为你实际的配置参数
    private static final String REGION = "集群所在区域";
    private static final String PROJECT_ID = "GCP项目ID";
    private static final String CLUSTER_NAME = "DataProc集群名称";
    private static final String SPARK_MAIN_CLASS = "Spark作业主类全路径";
    private static final String JAR_GS_URI = "Spark Jar包的GS地址";

    @Override
    public void service(HttpRequest request, HttpResponse response) throws IOException {
        String dataprocEndpoint = String.format("%s-dataproc.googleapis.com:443", REGION);
        JobControllerSettings controllerSettings = JobControllerSettings.newBuilder()
                .setEndpoint(dataprocEndpoint)
                .build();

        try (JobControllerClient controllerClient = JobControllerClient.create(controllerSettings)) {
            // 配置自定义JobId
            JobReference jobReference = JobReference.newBuilder()
                    .setProjectId(PROJECT_ID)
                    .setJobId("spark-job-" + UUID.randomUUID())
                    .build();
            
            // 配置作业参数
            JobPlacement placement = JobPlacement.newBuilder()
                    .setClusterName(CLUSTER_NAME)
                    .build();
            SparkJob sparkJob = SparkJob.newBuilder()
                    .setMainClass(SPARK_MAIN_CLASS)
                    .addJarFileUris(JAR_GS_URI)
                    .build();
            Job submitJob = Job.newBuilder()
                    .setReference(jobReference)
                    .setPlacement(placement)
                    .setSparkJob(sparkJob)
                    .build();
            
            // 异步提交作业
            Job resultJob = controllerClient.submitJobAsOperationAsync(PROJECT_ID, REGION, submitJob).get();
            
            // 返回作业提交结果
            response.setStatusCode(200);
            response.getWriter().write(String.format("作业提交成功，JobId：%s", resultJob.getReference().getJobId()));
        } catch (InterruptedException | ExecutionException e) {
            response.setStatusCode(500);
            response.getWriter().write(String.format("作业提交失败：%s", e.getMessage()));
            Thread.currentThread().interrupt();
        }
    }
}

部署后配置 #

云函数部署完成后，在IAM页面找到云函数的默认服务账号，为其授予Dataproc 作业提交者角色（对应权限名roles/dataproc.job.submitter）即可正常运行。

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