'Creates a job flow using the config from the EMR connection.
Keys of the json extra hash may have the arguments of the boto3 run_job_flow method.
Overrides for this config may be passed as the job_flow_overrides.'

def create_job_flow(self, job_flow_overrides):

    if (not self.emr_conn_id):
        raise AirflowException('emr_conn_id    must    be    present    to    use    create_job_flow')
    emr_conn = self.get_connection(self.emr_conn_id)
    config = emr_conn.extra_dejson.copy()
    config.update(job_flow_overrides)
    response = self.get_conn().run_job_flow(Name=config.get('Name'), LogUri=config.get('LogUri'), ReleaseLabel=config.get('ReleaseLabel'), Instances=config.get('Instances'), Steps=config.get('Steps', []), BootstrapActions=config.get('BootstrapActions', []), Applications=config.get('Applications'), Configurations=config.get('Configurations', []), VisibleToAllUsers=config.get('VisibleToAllUsers'), JobFlowRole=config.get('JobFlowRole'), ServiceRole=config.get('ServiceRole'), Tags=config.get('Tags'))
    return response
