public List<String> listActiveClusterNames() {
        AmazonElasticMapReduce emr = sparkEmrClientBuilder.build();
        List<ClusterSummary> csrl =
                emr.listClusters(new ListClustersRequest().withClusterStates(activeClusterStates)).getClusters();
        logClusters(csrl);
        List<String> res = new ArrayList<>(csrl.size());
        for (ClusterSummary csr : csrl) res.add(csr.getName());
        return res;
    }

    /**
     * List existing active cluster IDs
     *
     * @return cluster IDs
     */
    public List<String> listActiveClusterIds() {
        AmazonElasticMapReduce emr = sparkEmrClientBuilder.build();
        List<ClusterSummary> csrl =
                emr.listClusters(new ListClustersRequest().withClusterStates(activeClusterStates)).getClusters();
        logClusters(csrl);
        List<String> res = new ArrayList<>(csrl.size());
        for (ClusterSummary csr : csrl) res.add(csr.getId());
        return res;
    }


    /**
     * Terminates a cluster
     */
    public void terminateCluster() {
        AmazonElasticMapReduce emr = sparkEmrClientBuilder.build();
        Optional<ClusterSummary> optClusterSum = findClusterWithName(emr, sparkClusterName);
        if (!optClusterSum.isPresent()) {
            log.error(String.format("The cluster with name %s , requested for deletion, does not exist.", sparkClusterName));
        } else {
            String id = optClusterSum.get().getId();
            emr.terminateJobFlows((new TerminateJobFlowsRequest()).withJobFlowIds(id));
            log.info(String.format("The cluster with id %s is terminating.", id));
        }
    }

    // The actual job-sumission logic
    private void submitJob(AmazonElasticMapReduce emr, String mainClass, List<String> args, Map<String, String> sparkConfs, File uberJar) throws Exception {
        AmazonS3URI s3Jar = new AmazonS3URI(sparkS3JarFolder + "/" + uberJar.getName());
        log.info(String.format("Placing uberJar %s to %s", uberJar.getPath(), s3Jar.toString()));
        PutObjectRequest putRequest = sparkS3PutObjectDecorator.call(
                new PutObjectRequest(s3Jar.getBucket(), s3Jar.getKey(), uberJar)
        );
        sparkS3ClientBuilder.build().putObject(putRequest);
        // The order of these matters
        List<String> sparkSubmitArgs = Arrays.asList(
                "spark-submit",
                "--deploy-mode",
                "cluster",
                "--class",
                mainClass
        );
        for (Map.Entry<String, String> e : sparkConfs.entrySet()) {
            sparkSubmitArgs.add(String.format("--conf %s = %s ", e.getKey(), e.getValue()));
        }
        sparkSubmitArgs.add(s3Jar.toString());
        sparkSubmitArgs.addAll(args);
        StepConfig step = new StepConfig()
                .withActionOnFailure(ActionOnFailure.CONTINUE)
                .withName("Spark step")
                .withHadoopJarStep(
                        new HadoopJarStepConfig()
                                .withJar("command-runner.jar")
                                .withArgs(sparkSubmitArgs)
                );

        Optional<ClusterSummary> optCsr = findClusterWithName(emr, sparkClusterName);
        if (optCsr.isPresent()) {
            ClusterSummary csr = optCsr.get();
            emr.addJobFlowSteps(
                    new AddJobFlowStepsRequest()
                            .withJobFlowId(csr.getId())
                            .withSteps(step));
            log.info(
                    String.format("Your job is added to the cluster with id %s.", csr.getId())
            );
        } else {
            // If the cluster wasn't started, it's assumed ot be throwaway
            List<StepConfig> steps = sparkRunJobFlowRequest.getSteps();
            steps.add(step);
            RunJobFlowRequest jobFlowRequest = sparkRunJobFlowRequest
                    .withSteps(steps)
                    .withInstances(sparkJobFlowInstancesConfig.withKeepJobFlowAliveWhenNoSteps(false));

            RunJobFlowResult res = emr.runJobFlow(jobFlowRequest);
            log.info("Your new cluster's id is %s.", res.getJobFlowId());
        }

    }

    /**
     * Submit a Spark Job with a specified main class
     */
    public void sparkSubmitJobWithMain(String[] args, String mainClass, File uberJar) throws Exception {
        AmazonElasticMapReduce emr = sparkEmrClientBuilder.build();
        submitJob(emr, mainClass, Arrays.asList(args), sparkSubmitConfs, uberJar);
    }

    private void checkStatus(AmazonElasticMapReduce emr, String clusterId) throws InterruptedException {
        log.info(".");
        com.amazonaws.services.elasticmapreduce.model.Cluster dcr =
                emr.describeCluster((new DescribeClusterRequest()).withClusterId(clusterId)).getCluster();
        String state = dcr.getStatus().getState();
        long timeOutTime = System.currentTimeMillis() + ((long) sparkTimeoutDurationMinutes * 60 * 1000);

        Boolean activated = Arrays.asList(activeClusterStates).contains(ClusterState.fromValue(state));
        Boolean timedOut = System.currentTimeMillis() > timeOutTime;
        if (activated && timedOut) {
            emr.terminateJobFlows(
                    new TerminateJobFlowsRequest().withJobFlowIds(clusterId)
            );
            log.error("Timeout. Cluster terminated.");
        } else if (!activated) {
            Boolean hasAbnormalStep = false;
            StepSummary stepS = null;
            List<StepSummary> steps = emr.listSteps(new ListStepsRequest().withClusterId(clusterId)).getSteps();
            for (StepSummary step : steps) {
                if (step.getStatus().getState() != StepState.COMPLETED.toString()) {
                    hasAbnormalStep = true;
                    stepS = step;
                }
            }
            if (hasAbnormalStep && stepS != null)
                log.error(String.format("Cluster %s terminated with an abnormal step, name %s, id %s", clusterId, stepS.getName(), stepS.getId()));
            else
                log.info("Cluster %s terminated without error.", clusterId);
        } else {
            Thread.sleep(5000);
            checkStatus(emr, clusterId);
        }
    }

    /**
     * Monitor the cluster and terminates when it times out
     */
    public void sparkMonitor() throws InterruptedException {
        AmazonElasticMapReduce emr = sparkEmrClientBuilder.build();
        Optional<ClusterSummary> optCsr = findClusterWithName(emr, sparkClusterName);
        if (!optCsr.isPresent()) {
            log.error(String.format("The cluster with name %s does not exist.", sparkClusterName));
        } else {
            ClusterSummary csr = optCsr.get();
            log.info(String.format("found cluster with id %s, starting monitoring", csr.getId()));
            checkStatus(emr, csr.getId());
        }
    }

    @Data
    public static class Builder {

        protected String sparkClusterName = RandomStringUtils.randomAlphanumeric(12);
        protected String sparkAwsRegion = "us-east-1";
        protected String sparkEmrRelease = "emr-5.9.0";
        protected String sparkEmrServiceRole = "EMR_DefaultRole";
        protected List<EmrConfig> sparkEmrConfigs = Collections.emptyList();
        protected String sparkSubNetid = null;
        protected List<String> sparkSecurityGroupIds = Collections.emptyList();
        protected int sparkInstanceCount = 1;
        protected String sparkInstanceType = "m3.xlarge";
        protected Optional<Float> sparkInstanceBidPrice = Optional.empty();
        protected String sparkInstanceRole = "EMR_EC2_DefaultRole";
        protected String sparkS3JarFolder = "changeme";
        protected int sparkTimeoutDurationMinutes = 90;

        protected AmazonElasticMapReduceClientBuilder sparkEmrClientBuilder;
        protected AmazonS3ClientBuilder sparkS3ClientBuilder;
        protected JobFlowInstancesConfig sparkJobFlowInstancesConfig;
        protected RunJobFlowRequest sparkRunJobFlowRequest;

        // This should allow the user to decorate the put call to add metadata to the jar put command, such as security groups,
        protected Function<PutObjectRequest, PutObjectRequest> sparkS3PutObjectDecorator = new Function<PutObjectRequest, PutObjectRequest>() {
            @Override
            public PutObjectRequest call(PutObjectRequest putObjectRequest) throws Exception {
                return putObjectRequest;
            }
        };
        protected Map<String, String> sparkSubmitConfs;


        /**
         * Defines the EMR cluster's name
         *
         * @param clusterName the EMR cluster's name
         * @return an EMR cluster builder
         */
        public Builder clusterName(String clusterName) {
            this.sparkClusterName = clusterName;
            return this;
        }

        /**
         * Defines the EMR cluster's region
         * See https://docs.aws.amazon.com/general/latest/gr/rande.html
         *
         * @param region the EMR cluster's region
         * @return an EMR cluster builder
         */
        public Builder awsRegion(String region) {
            this.sparkAwsRegion = region;
            return this;
        }

        /**
         * Defines the EMR release version to be used in this cluster
         * uses a release label
         * See https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-4.2.0/emr-release-differences.html#emr-release-label
         *
         * @param releaseLabel the EMR release label
         * @return an EM cluster Builder
         */
        public Builder emrRelease(String releaseLabel) {
            this.sparkEmrRelease = releaseLabel;
            return this;
        }

        /**
         * Defines the IAM role to be assumed by the EMR service
         * <p>
         * https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-service.html
         *
         * @param serviceRole the service role
         * @return an EM cluster Builder
         */
        public Builder emrServiceRole(String serviceRole) {
            this.sparkEmrServiceRole = serviceRole;
            return this;
        }

        /**
         * A list of configuration parameters to apply to EMR instances.
         *
         * @param configs the EMR configurations to apply to this cluster
         * @return an EMR cluster builder
         */
        public Builder emrConfigs(List<EmrConfig> configs) {
            this.sparkEmrConfigs = configs;
            return this;
        }

        /**
         * The id of the EC2 subnet to be used for this Spark EMR service
         * see https://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/VPC_Subnets.html
         *
         * @param id the subnet ID
         * @return an EMR cluster builder
         */
        public Builder subnetId(String id) {
            this.sparkSubNetid = id;
            return this;
        }

        /**
         * The id of additional security groups this deployment should adopt for both master and slaves
         *
         * @param securityGroups
         * @return an EMR cluster builder
         */
        public Builder securityGroupIDs(List<String> securityGroups) {
            this.sparkSecurityGroupIds = securityGroups;
            return this;
        }

        /**
         * The number of instances this deployment should comprise of
         *
         * @param count the number of instances for this cluster
         * @rturn an EMR cluster buidler
         */
        public Builder instanceCount(int count) {
            this.sparkInstanceCount = count;
            return this;
        }

        /**
         * The type of instance this cluster should comprise of
         * See https://aws.amazon.com/ec2/instance-types/
         *
         * @param instanceType the type of instance for this cluster
         * @return an EMR cluster builder
         */
        public Builder instanceType(String instanceType) {
            this.sparkInstanceType = instanceType;
            return this;
        }

        /**
         * The optional bid value for this cluster's spot instances
         * see https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/how-spot-instances-work.html
         * Uses the on-demand market if empty.
         *
         * @param optBid the Optional bid price for this cluster's instnces
         * @return an EMR cluster Builder
         */
        public Builder instanceBidPrice(Optional<Float> optBid) {
            this.sparkInstanceBidPrice = optBid;
            return this;
        }

        /**
         * The EC2 instance role that this cluster's instances should assume
         * see https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html
         *
         * @param role the intended instance role
         * @return an EMR cluster builder
         */
        public Builder instanceRole(String role) {
            this.sparkInstanceRole = role;
            return this;
        }

        /**
         * the S3 folder in which to find the application jar
         *
         * @param jarfolder the S3 folder in which to find a jar
         * @return an EMR cluster builder
         */
        public Builder s3JarFolder(String jarfolder) {
            this.sparkS3JarFolder = jarfolder;
            return this;
        }

        /**
         * The timeout duration for this Spark EMR cluster, in minutes
         *
         * @param timeoutMinutes
         * @return an EMR cluster builder
         */
        public Builder sparkTimeOutDurationMinutes(int timeoutMinutes) {
            this.sparkTimeoutDurationMinutes = timeoutMinutes;
            return this;
        }

        /**
         * Creates an EMR Spark cluster deployment
         *
         * @return a SparkEMRClient
         */
        public SparkEMRClient build() {
            this.sparkEmrClientBuilder = AmazonElasticMapReduceClientBuilder.standard().withRegion(sparkAwsRegion);
            this.sparkS3ClientBuilder = AmazonS3ClientBuilder.standard().withRegion(sparkAwsRegion);
            // note this will be kept alive without steps, an arbitrary choice to avoid rapid test-teardown-restart cycles
            this.sparkJobFlowInstancesConfig = (new JobFlowInstancesConfig()).withKeepJobFlowAliveWhenNoSteps(true);
            if (this.sparkSubNetid != null)
                this.sparkJobFlowInstancesConfig = this.sparkJobFlowInstancesConfig.withEc2SubnetId(this.sparkSubNetid);
            if (!this.sparkSecurityGroupIds.isEmpty()) {
                this.sparkJobFlowInstancesConfig = this.sparkJobFlowInstancesConfig.withAdditionalMasterSecurityGroups(this.sparkSecurityGroupIds);
                this.sparkJobFlowInstancesConfig = this.sparkJobFlowInstancesConfig.withAdditionalSlaveSecurityGroups(this.sparkSecurityGroupIds);
            }

            InstanceGroupConfig masterConfig =
                    (new InstanceGroupConfig()).withInstanceCount(1).withInstanceRole(InstanceRoleType.MASTER).withInstanceType(sparkInstanceType);
            if (sparkInstanceBidPrice.isPresent()) {
                masterConfig = masterConfig.withMarket(MarketType.SPOT).withBidPrice(sparkInstanceBidPrice.get().toString());
            } else {
                masterConfig = masterConfig.withMarket(MarketType.ON_DEMAND);
            }

            int slaveCount = sparkInstanceCount - 1;
            InstanceGroupConfig slaveConfig =
                    (new InstanceGroupConfig()).withInstanceCount(slaveCount).withInstanceRole(InstanceRoleType.CORE).withInstanceRole(sparkInstanceType);
            if (sparkInstanceBidPrice.isPresent()) {
                slaveConfig = slaveConfig.withMarket(MarketType.SPOT).withBidPrice(sparkInstanceBidPrice.get().toString());
            } else {
                slaveConfig = slaveConfig.withMarket(MarketType.ON_DEMAND);
            }
            if (slaveCount > 0) {
                this.sparkJobFlowInstancesConfig = this.sparkJobFlowInstancesConfig.withInstanceGroups(Arrays.asList(masterConfig, slaveConfig));
            } else {
                this.sparkJobFlowInstancesConfig = this.sparkJobFlowInstancesConfig.withInstanceGroups(slaveConfig);
            }

            this.sparkRunJobFlowRequest = new RunJobFlowRequest();
            if (!sparkEmrConfigs.isEmpty()) {
                List<Configuration> emrConfigs = new ArrayList<>();
                for (EmrConfig config : sparkEmrConfigs) {
                    emrConfigs.add(config.toAwsConfig());
                }
                this.sparkRunJobFlowRequest = this.sparkRunJobFlowRequest.withConfigurations(emrConfigs);
            }
            this.sparkRunJobFlowRequest =
                    this.sparkRunJobFlowRequest.withName(sparkClusterName).withApplications((new Application()).withName("Spark"))
                            .withReleaseLabel(sparkEmrRelease)
                            .withServiceRole(sparkEmrServiceRole)
                            .withJobFlowRole(sparkInstanceRole)
                            .withInstances(this.sparkJobFlowInstancesConfig);

            return new SparkEMRClient(
                    this.sparkClusterName,
                    this.sparkAwsRegion,
                    this.sparkEmrRelease,
                    this.sparkEmrServiceRole,
                    this.sparkEmrConfigs,
                    this.sparkSubNetid,
                    this.sparkSecurityGroupIds,
                    this.sparkInstanceCount,
                    this.sparkInstanceType,
                    this.sparkInstanceBidPrice,
                    this.sparkInstanceRole,
                    this.sparkS3JarFolder,
                    this.sparkTimeoutDurationMinutes,
                    this.sparkEmrClientBuilder,
                    this.sparkS3ClientBuilder,
                    this.sparkJobFlowInstancesConfig,
                    this.sparkRunJobFlowRequest,
                    this.sparkS3PutObjectDecorator,
                    this.sparkSubmitConfs
            );
        }

    }

}
public List < String > listActiveClusterNames ( ) { AmazonElasticMapReduce emr = sparkEmrClientBuilder . build ( ) ; List < ClusterSummary > csrl = emr . listClusters ( new ListClustersRequest ( ) . withClusterStates ( activeClusterStates ) ) . getClusters ( ) ; logClusters ( csrl ) ; List < String > res = new ArrayList < > ( csrl . size ( ) ) ; for ( ClusterSummary csr : csrl ) res . add ( csr . getName ( ) ) ; return res ; } public List < String > listActiveClusterIds ( ) { AmazonElasticMapReduce emr = sparkEmrClientBuilder . build ( ) ; List < ClusterSummary > csrl = emr . listClusters ( new ListClustersRequest ( ) . withClusterStates ( activeClusterStates ) ) . getClusters ( ) ; logClusters ( csrl ) ; List < String > res = new ArrayList < > ( csrl . size ( ) ) ; for ( ClusterSummary csr : csrl ) res . add ( csr . getId ( ) ) ; return res ; } public void terminateCluster ( ) { AmazonElasticMapReduce emr = sparkEmrClientBuilder . build ( ) ; Optional < ClusterSummary > optClusterSum = findClusterWithName ( emr , sparkClusterName ) ; if ( ! optClusterSum . isPresent ( ) ) { log . error ( String . format ( "The cluster with name %s , requested for deletion, does not exist." , sparkClusterName ) ) ; } else { String id = optClusterSum . get ( ) . getId ( ) ; emr . terminateJobFlows ( ( new TerminateJobFlowsRequest ( ) ) . withJobFlowIds ( id ) ) ; log . info ( String . format ( "The cluster with id %s is terminating." , id ) ) ; } } private void submitJob ( AmazonElasticMapReduce emr , String mainClass , List < String > args , Map < String , String > sparkConfs , File uberJar ) throws Exception { AmazonS3URI s3Jar = new AmazonS3URI ( sparkS3JarFolder + "/" + uberJar . getName ( ) ) ; log . info ( String . format ( "Placing uberJar %s to %s" , uberJar . getPath ( ) , s3Jar . toString ( ) ) ) ; PutObjectRequest putRequest = sparkS3PutObjectDecorator . call ( new PutObjectRequest ( s3Jar . getBucket ( ) , s3Jar . getKey ( ) , uberJar ) ) ; sparkS3ClientBuilder . build ( ) . putObject ( putRequest ) ; List < String > sparkSubmitArgs = Arrays . asList ( "spark-submit" , "--deploy-mode" , "cluster" , "--class" , mainClass ) ; for ( Map . Entry < String , String > e : sparkConfs . entrySet ( ) ) { sparkSubmitArgs . add ( String . format ( "--conf %s = %s " , e . getKey ( ) , e . getValue ( ) ) ) ; } sparkSubmitArgs . add ( s3Jar . toString ( ) ) ; sparkSubmitArgs . addAll ( args ) ; StepConfig step = new StepConfig ( ) . withActionOnFailure ( ActionOnFailure . CONTINUE ) . withName ( "Spark step" ) . withHadoopJarStep ( new HadoopJarStepConfig ( ) . withJar ( "command-runner.jar" ) . withArgs ( sparkSubmitArgs ) ) ; Optional < ClusterSummary > optCsr = findClusterWithName ( emr , sparkClusterName ) ; if ( optCsr . isPresent ( ) ) { ClusterSummary csr = optCsr . get ( ) ; emr . addJobFlowSteps ( new AddJobFlowStepsRequest ( ) . withJobFlowId ( csr . getId ( ) ) . withSteps ( step ) ) ; log . info ( String . format ( "Your job is added to the cluster with id %s." , csr . getId ( ) ) ) ; } else { List < StepConfig > steps = sparkRunJobFlowRequest . getSteps ( ) ; steps . add ( step ) ; RunJobFlowRequest jobFlowRequest = sparkRunJobFlowRequest . withSteps ( steps ) . withInstances ( sparkJobFlowInstancesConfig . withKeepJobFlowAliveWhenNoSteps ( false ) ) ; RunJobFlowResult res = emr . runJobFlow ( jobFlowRequest ) ; log . info ( "Your new cluster's id is %s." , res . getJobFlowId ( ) ) ; } } public void sparkSubmitJobWithMain ( String [ ] args , String mainClass , File uberJar ) throws Exception { AmazonElasticMapReduce emr = sparkEmrClientBuilder . build ( ) ; submitJob ( emr , mainClass , Arrays . asList ( args ) , sparkSubmitConfs , uberJar ) ; } private void checkStatus ( AmazonElasticMapReduce emr , String clusterId ) throws InterruptedException { log . info ( "." ) ; com . amazonaws . services . elasticmapreduce . model . Cluster dcr = emr . describeCluster ( ( new DescribeClusterRequest ( ) ) . withClusterId ( clusterId ) ) . getCluster ( ) ; String state = dcr . getStatus ( ) . getState ( ) ; long timeOutTime = System . currentTimeMillis ( ) + ( ( long ) sparkTimeoutDurationMinutes * 60 * 1000 ) ; Boolean activated = Arrays . asList ( activeClusterStates ) . contains ( ClusterState . fromValue ( state ) ) ; Boolean timedOut = System . currentTimeMillis ( ) > timeOutTime ; if ( activated && timedOut ) { emr . terminateJobFlows ( new TerminateJobFlowsRequest ( ) . withJobFlowIds ( clusterId ) ) ; log . error ( "Timeout. Cluster terminated." ) ; } else if ( ! activated ) { Boolean hasAbnormalStep = false ; StepSummary stepS = null ; List < StepSummary > steps = emr . listSteps ( new ListStepsRequest ( ) . withClusterId ( clusterId ) ) . getSteps ( ) ; for ( StepSummary step : steps ) { if ( step . getStatus ( ) . getState ( ) != StepState . COMPLETED . toString ( ) ) { hasAbnormalStep = true ; stepS = step ; } } if ( hasAbnormalStep && stepS != null ) log . error ( String . format ( "Cluster %s terminated with an abnormal step, name %s, id %s" , clusterId , stepS . getName ( ) , stepS . getId ( ) ) ) ; else log . info ( "Cluster %s terminated without error." , clusterId ) ; } else { Thread . sleep ( 5000 ) ; checkStatus ( emr , clusterId ) ; } } public void sparkMonitor ( ) throws InterruptedException { AmazonElasticMapReduce emr = sparkEmrClientBuilder . build ( ) ; Optional < ClusterSummary > optCsr = findClusterWithName ( emr , sparkClusterName ) ; if ( ! optCsr . isPresent ( ) ) { log . error ( String . format ( "The cluster with name %s does not exist." , sparkClusterName ) ) ; } else { ClusterSummary csr = optCsr . get ( ) ; log . info ( String . format ( "found cluster with id %s, starting monitoring" , csr . getId ( ) ) ) ; checkStatus ( emr , csr . getId ( ) ) ; } } @ Data public static class Builder { protected String sparkClusterName = RandomStringUtils . randomAlphanumeric ( 12 ) ; protected String sparkAwsRegion = "us-east-1" ; protected String sparkEmrRelease = "emr-5.9.0" ; protected String sparkEmrServiceRole = "EMR_DefaultRole" ; protected List < EmrConfig > sparkEmrConfigs = Collections . emptyList ( ) ; protected String sparkSubNetid = null ; protected List < String > sparkSecurityGroupIds = Collections . emptyList ( ) ; protected int sparkInstanceCount = 1 ; protected String sparkInstanceType = "m3.xlarge" ; protected Optional < Float > sparkInstanceBidPrice = Optional . empty ( ) ; protected String sparkInstanceRole = "EMR_EC2_DefaultRole" ; protected String sparkS3JarFolder = "changeme" ; protected int sparkTimeoutDurationMinutes = 90 ; protected AmazonElasticMapReduceClientBuilder sparkEmrClientBuilder ; protected AmazonS3ClientBuilder sparkS3ClientBuilder ; protected JobFlowInstancesConfig sparkJobFlowInstancesConfig ; protected RunJobFlowRequest sparkRunJobFlowRequest ; protected Function < PutObjectRequest , PutObjectRequest > sparkS3PutObjectDecorator = new Function < PutObjectRequest , PutObjectRequest > ( ) { @ Override public PutObjectRequest call ( PutObjectRequest putObjectRequest ) throws Exception { return putObjectRequest ; } } ; protected Map < String , String > sparkSubmitConfs ; public Builder clusterName ( String clusterName ) { this . sparkClusterName = clusterName ; return this ; } public Builder awsRegion ( String region ) { this . sparkAwsRegion = region ; return this ; } public Builder emrRelease ( String releaseLabel ) { this . sparkEmrRelease = releaseLabel ; return this ; } public Builder emrServiceRole ( String serviceRole ) { this . sparkEmrServiceRole = serviceRole ; return this ; } public Builder emrConfigs ( List < EmrConfig > configs ) { this . sparkEmrConfigs = configs ; return this ; } public Builder subnetId ( String id ) { this . sparkSubNetid = id ; return this ; } public Builder securityGroupIDs ( List < String > securityGroups ) { this . sparkSecurityGroupIds = securityGroups ; return this ; } public Builder instanceCount ( int count ) { this . sparkInstanceCount = count ; return this ; } public Builder instanceType ( String instanceType ) { this . sparkInstanceType = instanceType ; return this ; } public Builder instanceBidPrice ( Optional < Float > optBid ) { this . sparkInstanceBidPrice = optBid ; return this ; } public Builder instanceRole ( String role ) { this . sparkInstanceRole = role ; return this ; } public Builder s3JarFolder ( String jarfolder ) { this . sparkS3JarFolder = jarfolder ; return this ; } public Builder sparkTimeOutDurationMinutes ( int timeoutMinutes ) { this . sparkTimeoutDurationMinutes = timeoutMinutes ; return this ; } public SparkEMRClient build ( ) { this . sparkEmrClientBuilder = AmazonElasticMapReduceClientBuilder . standard ( ) . withRegion ( sparkAwsRegion ) ; this . sparkS3ClientBuilder = AmazonS3ClientBuilder . standard ( ) . withRegion ( sparkAwsRegion ) ; this . sparkJobFlowInstancesConfig = ( new JobFlowInstancesConfig ( ) ) . withKeepJobFlowAliveWhenNoSteps ( true ) ; if ( this . sparkSubNetid != null ) this . sparkJobFlowInstancesConfig = this . sparkJobFlowInstancesConfig . withEc2SubnetId ( this . sparkSubNetid ) ; if ( ! this . sparkSecurityGroupIds . isEmpty ( ) ) { this . sparkJobFlowInstancesConfig = this . sparkJobFlowInstancesConfig . withAdditionalMasterSecurityGroups ( this . sparkSecurityGroupIds ) ; this . sparkJobFlowInstancesConfig = this . sparkJobFlowInstancesConfig . withAdditionalSlaveSecurityGroups ( this . sparkSecurityGroupIds ) ; } InstanceGroupConfig masterConfig = ( new InstanceGroupConfig ( ) ) . withInstanceCount ( 1 ) . withInstanceRole ( InstanceRoleType . MASTER ) . withInstanceType ( sparkInstanceType ) ; if ( sparkInstanceBidPrice . isPresent ( ) ) { masterConfig = masterConfig . withMarket ( MarketType . SPOT ) . withBidPrice ( sparkInstanceBidPrice . get ( ) . toString ( ) ) ; } else { masterConfig = masterConfig . withMarket ( MarketType . ON_DEMAND ) ; } int slaveCount = sparkInstanceCount - 1 ; InstanceGroupConfig slaveConfig = ( new InstanceGroupConfig ( ) ) . withInstanceCount ( slaveCount ) . withInstanceRole ( InstanceRoleType . CORE ) . withInstanceRole ( sparkInstanceType ) ; if ( sparkInstanceBidPrice . isPresent ( ) ) { slaveConfig = slaveConfig . withMarket ( MarketType . SPOT ) . withBidPrice ( sparkInstanceBidPrice . get ( ) . toString ( ) ) ; } else { slaveConfig = slaveConfig . withMarket ( MarketType . ON_DEMAND ) ; } if ( slaveCount > 0 ) { this . sparkJobFlowInstancesConfig = this . sparkJobFlowInstancesConfig . withInstanceGroups ( Arrays . asList ( masterConfig , slaveConfig ) ) ; } else { this . sparkJobFlowInstancesConfig = this . sparkJobFlowInstancesConfig . withInstanceGroups ( slaveConfig ) ; } this . sparkRunJobFlowRequest = new RunJobFlowRequest ( ) ; if ( ! sparkEmrConfigs . isEmpty ( ) ) { List < Configuration > emrConfigs = new ArrayList < > ( ) ; for ( EmrConfig config : sparkEmrConfigs ) { emrConfigs . add ( config . toAwsConfig ( ) ) ; } this . sparkRunJobFlowRequest = this . sparkRunJobFlowRequest . withConfigurations ( emrConfigs ) ; } this . sparkRunJobFlowRequest = this . sparkRunJobFlowRequest . withName ( sparkClusterName ) . withApplications ( ( new Application ( ) ) . withName ( "Spark" ) ) . withReleaseLabel ( sparkEmrRelease ) . withServiceRole ( sparkEmrServiceRole ) . withJobFlowRole ( sparkInstanceRole ) . withInstances ( this . sparkJobFlowInstancesConfig ) ; return new SparkEMRClient ( this . sparkClusterName , this . sparkAwsRegion , this . sparkEmrRelease , this . sparkEmrServiceRole , this . sparkEmrConfigs , this . sparkSubNetid , this . sparkSecurityGroupIds , this . sparkInstanceCount , this . sparkInstanceType , this . sparkInstanceBidPrice , this . sparkInstanceRole , this . sparkS3JarFolder , this . sparkTimeoutDurationMinutes , this . sparkEmrClientBuilder , this . sparkS3ClientBuilder , this . sparkJobFlowInstancesConfig , this . sparkRunJobFlowRequest , this . sparkS3PutObjectDecorator , this . sparkSubmitConfs ) ; } } }
Lists existing active clusters Names

@return cluster names