在单个 EMR 集群中调用多个 Spark 作业

Question

我想在单个 EMR 集群中使用 spark-submit 调用多个 spark 作业。 EMR 支持吗？ 如何做到这一点？ 我现在使用 AWS Lambda 为我的 spark 作业调用 EMR 作业，但我们希望扩展到单个 EMR 集群中的多个 spark 作业。

Answer 1

您可以在一个 EMR 上按顺序运行多个 Spark 作业 - 也就是说，下一个作业将在前一个作业完成后启动。这是使用 EMR 步骤完成的。

我使用 Java SDK 来运行它，但您可以在此文档中看到 how to add step using CLI only。

我下面的代码使用 spark-submit，但它不像在 CLI 中那样直接运行。相反，我将它作为 shell 脚本运行，并为 HADOOP_USER_NAME 包含了一个环境变量，因此 spark 作业在我指定的用户名下运行。如果您想使用您登录 EMR 的用户名（默认情况下为 hadoop）运行作业，则可以跳过它。

在下面的代码摘录中，对象emr 的类型为AmazonElasticMapReduce，在sdk 中提供。如果您使用的是 CLI 方法，则不需要它。

uploadConfFile 等一些辅助方法是不言自明的。我为 spark 应用程序使用了广泛的配置，与 files 和 jars 可以在本地或在 s3/hdfs 中不同，配置文件必须在 EMR 本身的本地文件中。

完成后，您将在 EMR 集群上创建一个步骤，该步骤将启动一个新的 spark 应用程序。您可以在 EMR 上指定多个步骤，这些步骤将一个接一个地运行。

//Upload the spark configuration you wish to use to a local file    
uploadConfFile(clusterId, sparkConf, confFileName);

//create a list of arguments - which is the complete command for spark-submit
List<String> stepargs = new ArrayList<String>();
//start with an envelope to specify the hadoop user name
stepargs.add("/bin/sh");
stepargs.add("-c");
//call to spark-submit with the incantation stating its arguments are provided next.
stepargs.add("HADOOP_USER_NAME="+task.getUserName()+" spark-submit \"$@\"");
stepargs.add("sh");
//add the spark-submit arguments
stepargs.add("--class");
stepargs.add(mainClass);
stepargs.add("--deploy-mode");
stepargs.add("cluster");
stepargs.add("--master");
stepargs.add("yarn");
stepargs.add("--files");
//a comma-separated list of file paths in s3
stepargs.add(files);
stepargs.add("--jars");
//a comma-separated list of file paths in s3
stepargs.add(jars);
stepargs.add("--properties-file");
//the file we uploaded to the EMR, with its full path
stepargs.add(confFileName);
stepargs.add(jar);
//add the jar specific arguments in here

AddJobFlowStepsResult result = emr.addJobFlowSteps(new AddJobFlowStepsRequest()
    .withJobFlowId(clusterId)
    .withSteps(new StepConfig()
            .withName(name)
            .withActionOnFailure(ActionOnFailure.CONTINUE)
            .withHadoopJarStep(new HadoopJarStepConfig()
                    .withJar("command-runner.jar")
                    .withArgs(stepargs))));