run.py 文件源码

def run_cluster(self, cluster_configuration):
        """
        Launch a cluster based on given configuration
        :param cluster_configuration: string
        """
        aws_arguments = self.__job_flow_configuration.convert_to_arguments(cluster_configuration)

        self.__logger.debug('Launching cluster with given configuration')
        response = self.__aws_emr_client.run_job_flow(**aws_arguments)
        self.__logger.info('Cluster launched with ID: {cluster_id}'.format(cluster_id=response['JobFlowId']))
        try:
            waiter = self.__aws_emr_client.get_waiter('cluster_running')
            self.__logger.info('Waiting for cluster to be ready')
            waiter.wait(ClusterId=response['JobFlowId'])
            self.__logger.info('Cluster up and running ID: {cluster_id}'.format(cluster_id=response['JobFlowId']))
        except WaiterError:
            self.__logger.error('Cluster creation failed')