使用 Amazon SDK for Java 创建 Amazon EMR 集群

Amazon SDK for Java 提供具有 Amazon EMR 功能的三个包：

有关这些包的更多信息，请参阅 Amazon SDK for Java API 参考。

以下示例说明了 SDK 如何使用 Amazon EMR 简化编程过程。下面的代码示例使用 StepFactory 对象（用于创建通用 Amazon EMR 步骤类型的帮助程序类）创建一个启用了调试的交互式 Hive 集群。

import com.amazonaws.AmazonClientException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduce;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClientBuilder;
import com.amazonaws.services.elasticmapreduce.model.*;
import com.amazonaws.services.elasticmapreduce.util.StepFactory;

public class Main {

	public static void main(String[] args) {
		AWSCredentialsProvider profile = null;		
		try {
			credentials_profile = new ProfileCredentialsProvider("default"); // specifies any named profile in .aws/credentials as the credentials provider
        } catch (Exception e) {
            throw new AmazonClientException(
                    "Cannot load credentials from .aws/credentials file. " +
                    "Make sure that the credentials file exists and that the profile name is defined within it.",
                    e);
        }
		
		// create an EMR client using the credentials and region specified in order to create the cluster
		AmazonElasticMapReduce emr = AmazonElasticMapReduceClientBuilder.standard()
			.withCredentials(credentials_profile)
			.withRegion(Regions.US_WEST_1)
			.build();
        
        // create a step to enable debugging in the AWS Management Console
		StepFactory stepFactory = new StepFactory(); 
		StepConfig enabledebugging = new StepConfig()
   			.withName("Enable debugging")
   			.withActionOnFailure("TERMINATE_JOB_FLOW")
   			.withHadoopJarStep(stepFactory.newEnableDebuggingStep());
        
        // specify applications to be installed and configured when EMR creates the cluster
		Application hive = new Application().withName("Hive");
		Application spark = new Application().withName("Spark");
		Application ganglia = new Application().withName("Ganglia");
		Application zeppelin = new Application().withName("Zeppelin");
		
		// create the cluster
		RunJobFlowRequest request = new RunJobFlowRequest()
	       		.withName("MyClusterCreatedFromJava")
	       		.withReleaseLabel("emr-5.20.0") // specifies the EMR release version label, we recommend the latest release
	       		.withSteps(enabledebugging)
	       		.withApplications(hive,spark,ganglia,zeppelin)
	       		.withLogUri("s3://path/to/my/emr/logs") // a URI in S3 for log files is required when debugging is enabled
	       		.withServiceRole("EMR_DefaultRole") // replace the default with a custom IAM service role if one is used
	       		.withJobFlowRole("EMR_EC2_DefaultRole") // replace the default with a custom EMR role for the EC2 instance profile if one is used
	       		.withInstances(new JobFlowInstancesConfig()
	       	   		.withEc2SubnetId("subnet-12ab34c56")
	           		.withEc2KeyName("myEc2Key") 
	           		.withInstanceCount(3) 
	           		.withKeepJobFlowAliveWhenNoSteps(true)    
	           		.withMasterInstanceType("m4.large")
	           		.withSlaveInstanceType("m4.large"));

	   RunJobFlowResult result = emr.runJobFlow(request);  
	   System.out.println("The cluster ID is " + result.toString());

	}

}

至少，您必须传递分别对应于 EMR_DefaultRole 和 EMR_EC2_DefaultRole 的服务角色和任务流角色。可以通过为同一个账户调用此 Amazon CLI 命令来实现此目的。首先，查看这两个角色是否已存在：

aws iam list-roles | grep EMR

将显示实例配置文件 (EMR_EC2_DefaultRole) 和服务角色 (EMR_DefaultRole) (如果存在) ：

"RoleName": "EMR_DefaultRole", 
            "Arn": "arn:aws:iam::AccountID:role/EMR_DefaultRole"
            "RoleName": "EMR_EC2_DefaultRole", 
            "Arn": "arn:aws:iam::AccountID:role/EMR_EC2_DefaultRole"

如果默认角色不存在，则可以使用以下命令创建它们：

aws emr create-default-roles