import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
/**
 * Collect data for ElasticMapReduce.
 *
 * @param stats
 *            current statistics object.
 * @param account
 *            currently used credentials object.
 * @param region
 *            currently used aws region.
 */
public static void scanElasticMapReduce(AwsStats stats, AwsAccount account, Regions region) {
	LOG.debug("Scan for MapReduce in region " + region.getName() + " in account " + account.getAccountId());

	try {
		AmazonElasticMapReduce elasticMapReduce = new AmazonElasticMapReduceClient(account.getCredentials());
		elasticMapReduce.setRegion(Region.getRegion(region));

		List<ClusterSummary> list = elasticMapReduce.listClusters().getClusters();

		int totalItems = list.size();
		for (ClusterSummary cs : list) {
			stats.add(new AwsResource(cs.getName(), account.getAccountId(), AwsResourceType.ElasticMapReduce, region));
		}

		LOG.info(totalItems + " ElasticMapReduce clusters in region " + region.getName() + " in account " + account.getAccountId());
	} catch (AmazonServiceException ase) {
		if (ase.getErrorCode().contains("AccessDenied")) {
			LOG.info("Access denied for ElasticMapReduce in region " + region.getName() + " in account " + account.getAccountId());
		} else {
			LOG.error("Exception of ElasticMapReduce: " + ase.getMessage());
		}
	}
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
/**
 * Creates a client for accessing Amazon EMR service.
 *
 * @param awsParamsDto the AWS related parameters DTO that includes optional AWS credentials and proxy information
 *
 * @return the Amazon EMR client
 */
@Cacheable(DaoSpringModuleConfig.HERD_CACHE_NAME)
public AmazonElasticMapReduceClient getEmrClient(AwsParamsDto awsParamsDto)
{
    // Get client configuration.
    ClientConfiguration clientConfiguration = awsHelper.getClientConfiguration(awsParamsDto);

    // If specified, use the AWS credentials passed in.
    if (StringUtils.isNotBlank(awsParamsDto.getAwsAccessKeyId()))
    {
        return new AmazonElasticMapReduceClient(
            new BasicSessionCredentials(awsParamsDto.getAwsAccessKeyId(), awsParamsDto.getAwsSecretKey(), awsParamsDto.getSessionToken()),
            clientConfiguration);
    }
    // Otherwise, use the default AWS credentials provider chain.
    else
    {
        return new AmazonElasticMapReduceClient(clientConfiguration);
    }
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
public void monitorEMRStep() throws Exception {
	List<String> stepIds = new ArrayList<String>();
	Connection conn  = new com.mysql.jdbc.Driver().connect(props.getProperty("url"), props);
	ResultSet openStepsRS = conn.createStatement().executeQuery(props.getProperty("sql.retrieveOpenSteps"));
	AmazonElasticMapReduceClient emr = new AmazonElasticMapReduceClient();
	DescribeStepRequest stepReq=new  DescribeStepRequest();
	PreparedStatement ps = conn.prepareStatement(props.getProperty("sql.updateStepStatus"));
	while(openStepsRS.next()){
		
		stepReq.setClusterId(openStepsRS.getString("cluster_id"));
		stepReq.setStepId(openStepsRS.getString("step_id"));
		String stepState = emr.describeStep(stepReq).getStep().getStatus().getState();
		
			if(stepState.equals(StepState.COMPLETED.toString())){
				ps.setString(1,StepState.COMPLETED.toString());
			}else if (stepState.equals(StepState.FAILED.toString())){
				ps.setString(1,StepState.FAILED.toString());					
			}
			ps.setString(2,openStepsRS.getString("job_config_id"));
			ps.addBatch();
	}
	
	ps.executeBatch();
	ps.close();
	conn.close();
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
protected List<String> getActiveTaggedClusters() throws Exception{
	AmazonElasticMapReduceClient emrClient = new AmazonElasticMapReduceClient();
	List<String> waitingClusters = new ArrayList<String>();
	ListClustersResult clusterResult = emrClient.listClusters(new ListClustersRequest().withClusterStates(ClusterState.WAITING));
	
	DescribeClusterRequest specifcTagDescribe = new DescribeClusterRequest();
	specifcTagDescribe.putCustomQueryParameter("Cluster.Tags",null);
	 for( ClusterSummary cluster : clusterResult.getClusters()){
		 	System.out.println("list cluster id "+cluster.getId());
		 	List<Tag> tagList = emrClient.describeCluster(specifcTagDescribe.withClusterId(cluster.getId())).getCluster().getTags();
		 	for(Tag tag:tagList){
		 		if(tag.getKey().equals(props.getProperty("edba.cluster.tag.key"))){
		 			waitingClusters.add(cluster.getId());
		 		}
		 	}
		 	
	}
	return waitingClusters;
	
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
/**
 * Run the job.
 * @return a the JobFlowId of the job
 */
public String runJob() {

  // Get the credentials
  final AWSCredentials credentials =
      new BasicAWSCredentials(this.AWSAccessKey, this.AWSSecretKey);

  // Create the Amazon Elastic MapReduce object
  this.elasticMapReduceClient = new AmazonElasticMapReduceClient(credentials);

  // Set the end point
  this.elasticMapReduceClient.setEndpoint(this.endpoint);

  this.runFlowResult =
      this.elasticMapReduceClient.runJobFlow(this.runFlowRequest);

  return this.runFlowResult.getJobFlowId();
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
/**
 * Terminate EMR cluster, overrides terminate protection if requested.
 */
@Override
public void terminateEmrCluster(AmazonElasticMapReduceClient emrClient, String clusterId, boolean overrideTerminationProtection)
{
    // Override terminate protection if requested.
    if (overrideTerminationProtection)
    {
        // Set termination protection
        emrClient.setTerminationProtection(new SetTerminationProtectionRequest().withJobFlowIds(clusterId).withTerminationProtected(false));
    }

    // Terminate the job flow
    emrClient.terminateJobFlows(new TerminateJobFlowsRequest().withJobFlowIds(clusterId));
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Test
public void testGetEmrClientCacheHitMiss()
{
    // Create an AWS parameters DTO that contains both AWS credentials and proxy information.
    AwsParamsDto awsParamsDto =
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT);

    // Get an Amazon EMR client.
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = awsClientFactory.getEmrClient(awsParamsDto);

    // Confirm a cache hit.
    assertEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT)));

    // Confirm a cache miss due to AWS credentials.
    assertNotEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY_2, AWS_ASSUMED_ROLE_SECRET_KEY_2, AWS_ASSUMED_ROLE_SESSION_TOKEN_2, HTTP_PROXY_HOST,
            HTTP_PROXY_PORT)));

    // Confirm a cache miss due to http proxy information.
    assertNotEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST_2, HTTP_PROXY_PORT_2)));

    // Clear the cache.
    cacheManager.getCache(DaoSpringModuleConfig.HERD_CACHE_NAME).clear();

    // Confirm a cache miss due to cleared cache.
    assertNotEquals(amazonElasticMapReduceClient, awsClientFactory.getEmrClient(awsParamsDto));
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Test
public void getEmrClientAssertClientConfigurationSet() throws Exception
{
    String httpProxyHost = "httpProxyHost";
    Integer httpProxyPort = 1234;

    AwsParamsDto awsParamsDto = new AwsParamsDto();
    awsParamsDto.setHttpProxyHost(httpProxyHost);
    awsParamsDto.setHttpProxyPort(httpProxyPort);
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = emrDao.getEmrClient(awsParamsDto);
    ClientConfiguration clientConfiguration = (ClientConfiguration) ReflectionTestUtils.getField(amazonElasticMapReduceClient, "clientConfiguration");
    assertNotNull(clientConfiguration);
    assertEquals(httpProxyHost, clientConfiguration.getProxyHost());
    assertEquals(httpProxyPort.intValue(), clientConfiguration.getProxyPort());
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Test
public void getEmrClientAssertClientConfigurationNotSetWhenProxyHostIsBlank() throws Exception
{
    String httpProxyHost = "";
    Integer httpProxyPort = 1234;

    AwsParamsDto awsParamsDto = new AwsParamsDto();
    awsParamsDto.setHttpProxyHost(httpProxyHost);
    awsParamsDto.setHttpProxyPort(httpProxyPort);
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = emrDao.getEmrClient(awsParamsDto);
    ClientConfiguration clientConfiguration = (ClientConfiguration) ReflectionTestUtils.getField(amazonElasticMapReduceClient, "clientConfiguration");
    assertNotNull(clientConfiguration);
    assertNull(clientConfiguration.getProxyHost());
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Test
public void getEmrClientAssertClientConfigurationNotSetWhenProxyPortIsNull() throws Exception
{
    String httpProxyHost = "httpProxyHost";
    Integer httpProxyPort = null;

    AwsParamsDto awsParamsDto = new AwsParamsDto();
    awsParamsDto.setHttpProxyHost(httpProxyHost);
    awsParamsDto.setHttpProxyPort(httpProxyPort);
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = emrDao.getEmrClient(awsParamsDto);
    ClientConfiguration clientConfiguration = (ClientConfiguration) ReflectionTestUtils.getField(amazonElasticMapReduceClient, "clientConfiguration");
    assertNotNull(clientConfiguration);
    assertNull(clientConfiguration.getProxyHost());
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Override
public ListClustersResult listEmrClusters(AmazonElasticMapReduceClient emrClient, ListClustersRequest listClustersRequest)
{
    List<ClusterSummary> clusterSummaryList = new ArrayList<>();
    for (MockEmrJobFlow cluster : emrClusters.values())
    {
        if (!listClustersRequest.getClusterStates().isEmpty() && listClustersRequest.getClusterStates().contains(cluster.getStatus()))
        {
            ClusterSummary clusterSummary = new ClusterSummary();
            clusterSummary.withId(cluster.getJobFlowId()).withName(cluster.getJobFlowName()).withStatus(new ClusterStatus().withState(cluster.getStatus())
                .withStateChangeReason(new ClusterStateChangeReason().withCode(cluster.getStatusChangeReason().getCode())
                    .withMessage(cluster.getStatusChangeReason().getMessage())).withTimeline(new ClusterTimeline().withCreationDateTime(
                    cluster.getStatusTimeline().getCreationTime() != null ? cluster.getStatusTimeline().getCreationTime().toGregorianCalendar().getTime() :
                        null).withEndDateTime(
                    cluster.getStatusTimeline().getEndTime() != null ? cluster.getStatusTimeline().getEndTime().toGregorianCalendar().getTime() : null)
                    .withReadyDateTime(
                        cluster.getStatusTimeline().getReadyTime() != null ? cluster.getStatusTimeline().getReadyTime().toGregorianCalendar().getTime() :
                            null)));
            clusterSummaryList.add(clusterSummary);
        }
    }
    if (StringUtils.isBlank(listClustersRequest.getMarker()))
    {
        return new ListClustersResult().withClusters(clusterSummaryList).withMarker(MOCK_EMR_MAKER);
    }
    else
    {
        return new ListClustersResult().withClusters(clusterSummaryList);
    }
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Override
public ListInstancesResult listClusterInstancesRequest(AmazonElasticMapReduceClient emrClient, ListInstancesRequest listInstancesRequest)
{
    MockEmrJobFlow cluster =
        getClusterByName(buildEmrClusterName(AbstractDaoTest.NAMESPACE, AbstractDaoTest.EMR_CLUSTER_DEFINITION_NAME, MOCK_CLUSTER_NOT_PROVISIONED_NAME));

    if (cluster != null && listInstancesRequest.getClusterId().equals(cluster.getJobFlowId()))
    {
        return new ListInstancesResult();
    }
    Instance instance = new Instance().withEc2InstanceId("EC2_EMR_MASTER_INSTANCE").withPrivateIpAddress("INSTANCE_IP_ADDRESS");
    return new ListInstancesResult().withInstances(instance);
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Override
public void terminateEmrCluster(AmazonElasticMapReduceClient emrClient, String clusterId, boolean overrideTerminationProtection)
{
    MockEmrJobFlow cluster = getClusterById(clusterId);
    if (cluster.getJobFlowName().endsWith(MockAwsOperationsHelper.AMAZON_SERVICE_EXCEPTION))
    {
        throw new AmazonServiceException(MockAwsOperationsHelper.AMAZON_SERVICE_EXCEPTION);
    }
    cluster.setStatus(ClusterState.TERMINATED.toString());
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Override
public ListInstanceFleetsResult listInstanceFleets(AmazonElasticMapReduceClient emrClient, ListInstanceFleetsRequest listInstanceFleetsRequest)
{
    ListInstanceFleetsResult listInstanceFleetsResult = new ListInstanceFleetsResult();
    List<InstanceFleet> instanceFleets = new ArrayList<>();
    InstanceFleet instanceFleet = new InstanceFleet();
    instanceFleet.setId("mock_instance_id_1");
    instanceFleet.setName("mock_instance_name");
    instanceFleets.add(instanceFleet);
    listInstanceFleetsResult.setInstanceFleets(instanceFleets);
    return listInstanceFleetsResult;
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Test
public void testGetListInstanceFleetsResult()
{
    // Create an AWS parameters DTO.
    AwsParamsDto awsParamsDto =
        new AwsParamsDto(AWS_ASSUMED_ROLE_ACCESS_KEY, AWS_ASSUMED_ROLE_SECRET_KEY, AWS_ASSUMED_ROLE_SESSION_TOKEN, HTTP_PROXY_HOST, HTTP_PROXY_PORT);

    // Create a mock AmazonElasticMapReduceClient.
    AmazonElasticMapReduceClient amazonElasticMapReduceClient = mock(AmazonElasticMapReduceClient.class);

    // Create a list instance fleets request.
    ListInstanceFleetsRequest listInstanceFleetsRequest = new ListInstanceFleetsRequest().withClusterId(EMR_CLUSTER_ID);

    // Create a list instance fleets result.
    ListInstanceFleetsResult listInstanceFleetsResult = new ListInstanceFleetsResult().withMarker(MARKER);

    // Mock the external calls.
    when(awsClientFactory.getEmrClient(awsParamsDto)).thenReturn(amazonElasticMapReduceClient);
    when(emrOperations.listInstanceFleets(amazonElasticMapReduceClient, listInstanceFleetsRequest)).thenReturn(listInstanceFleetsResult);

    // Call the method under test.
    ListInstanceFleetsResult result = emrDaoImpl.getListInstanceFleetsResult(EMR_CLUSTER_ID, awsParamsDto);

    // Verify the external calls.
    verify(awsClientFactory).getEmrClient(awsParamsDto);
    verify(emrOperations).listInstanceFleets(amazonElasticMapReduceClient, listInstanceFleetsRequest);
    verifyNoMoreInteractionsHelper();

    // Validate the results.
    assertEquals(listInstanceFleetsResult, result);
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
/**
 * Helper class to create and Amazon EMR cluster with HBase installed on that cluster
 * 
 * @param clusterIdentifier - cluster id if one exists
 * @param clusterName - name you want associated with this cluster
 * @param amiVersion - version of AMI that you wish to use for your HBase cluster
 * @param keypair - you need a keypair to SSH into the cluster
 * @param masterInstanceType - Amazon EC2 instance type for your master node
 * @param coreInstanceType - Amazon Ec2 instance tyoe for your core nodes
 * @param logUri - Specify a bucket for your EMR logs
 * @param numberOfNodes - total number of nodes in your cluster including the master node
 */

private void createEMRCluster(String clusterIdentifier,
		String clusterName,
		String amiVersion,
		String keypair,
		String masterInstanceType,
		String coreInstanceType,
		String logUri,
		int numberOfNodes) {
	// Make sure the EMR cluster is available
	AmazonElasticMapReduceClient emrClient = new AmazonElasticMapReduceClient(config.AWS_CREDENTIALS_PROVIDER);
	emrClient.setEndpoint(config.EMR_ENDPOINT);
	String clusterid = clusterIdentifier;
	if (!EMRUtils.clusterExists(emrClient, clusterIdentifier)) {
		clusterid = EMRUtils.createCluster(emrClient,
					clusterIdentifier,
					amiVersion,
					keypair,
					masterInstanceType,
					coreInstanceType,
					logUri,
					numberOfNodes);
	}
	// Update the emr cluster id and public DNS properties
	config.EMR_CLUSTER_IDENTIFIER = clusterid;
	config.EMR_CLUSTER_PUBLIC_DNS = EMRUtils.getPublicDns(emrClient, clusterid);			
	//make sure table exists
	if (!HBaseUtils.tablesExists(config.HBASE_TABLE_NAME, config.EMR_CLUSTER_PUBLIC_DNS, config.HBASE_REST_PORT)){
		HBaseUtils.createTable(config.HBASE_TABLE_NAME, config.EMR_CLUSTER_PUBLIC_DNS, config.HBASE_REST_PORT);	
	}
	
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
public HBaseEmitter(EMRHBaseKinesisConnectorConfiguration configuration) {
	// DynamoDB Config
    this.emrEndpoint = configuration.EMR_ENDPOINT;
    this.hbaseTableName = configuration.HBASE_TABLE_NAME;
    this.hbaseRestPort = configuration.HBASE_REST_PORT;
    this.emrPublicDns = configuration.EMR_CLUSTER_PUBLIC_DNS;
    // Client
    this.emrClient = new AmazonElasticMapReduceClient(configuration.AWS_CREDENTIALS_PROVIDER);
    this.emrClient.setEndpoint(this.emrEndpoint);
	LOG.info("EMRHBaseEmitter.....");
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
protected String fireEMRJob(String paramsStr,String clusterId){
	StepFactory stepFactory = new StepFactory();
	AmazonElasticMapReduceClient emr = new AmazonElasticMapReduceClient();
	emr.setRegion(Region.getRegion(Regions.fromName(System.getenv().get("AWS_REGION"))));
	Application sparkConfig = new Application()
			.withName("Spark");
	
	String[] params = paramsStr.split(",");
	StepConfig enabledebugging = new StepConfig()
			.withName("Enable debugging")
			.withActionOnFailure("TERMINATE_JOB_FLOW")
			.withHadoopJarStep(stepFactory.newEnableDebuggingStep());
	
	HadoopJarStepConfig sparkStepConf = new HadoopJarStepConfig()
			.withJar("command-runner.jar")
			.withArgs(params);	
	
	final StepConfig sparkStep = new StepConfig()
			.withName("Spark Step")
			.withActionOnFailure("CONTINUE")
			.withHadoopJarStep(sparkStepConf);

	
	AddJobFlowStepsRequest request = new AddJobFlowStepsRequest(clusterId)
			.withSteps(new ArrayList<StepConfig>(){{add(sparkStep);}});
			

	AddJobFlowStepsResult result = emr.addJobFlowSteps(request);
	return result.getStepIds().get(0);
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
@Override
public AmazonElasticMapReduceClient getEmrClient(AwsParamsDto awsParamsDto)
{
    return awsClientFactory.getEmrClient(awsParamsDto);
}
 

import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClient; //导入依赖的package包/类
/**
 * List EMR cluster instances
 */
@Override
public ListInstancesResult listClusterInstancesRequest(AmazonElasticMapReduceClient emrClient, ListInstancesRequest listInstancesRequest)
{
    return emrClient.listInstances(listInstancesRequest);
}