'Launches a CloudML job and wait for it to reach a terminal state.
:param project_id: The Google Cloud project id within which CloudML
job will be launched.
:type project_id: string
:param job: CloudML Job object that should be provided to the CloudML
API, such as:
\'jobId\': \'my_job_id\',
\'trainingInput\': {
\'scaleTier\': \'STANDARD_1\',
:type job: dict
:param use_existing_job_fn: In case that a CloudML job with the same
job_id already exist, this method (if provided) will decide whether
we should use this existing job, continue waiting for it to finish
and returning the job object. It should accepts a CloudML job
object, and returns a boolean value indicating whether it is OK to
reuse the existing job. If \'use_existing_job_fn\' is not provided,
we by default reuse the existing CloudML job.
:type use_existing_job_fn: function
:return: The CloudML job object if the job successfully reach a
terminal state (which might be FAILED or CANCELLED state).
:rtype: dict'

def create_job(self, project_id, job, use_existing_job_fn=None):

    request = self._cloudml.projects().jobs().create(parent='projects/{}'.format(project_id), body=job)
    job_id = job['jobId']
    try:
        request.execute()
    except errors.HttpError as e:
        if (e.resp.status == 409):
            if (use_existing_job_fn is not None):
                existing_job = self._get_job(project_id, job_id)
                if (not use_existing_job_fn(existing_job)):
                    logging.error('Job    with    job_id    {}    already    exist,    but    it    does    not    match    our    expectation:    {}'.format(job_id, existing_job))
                    raise
            logging.info('Job    with    job_id    {}    already    exist.    Will    waiting    for    it    to    finish'.format(job_id))
        else:
            logging.error('Failed    to    create    CloudML    job:    {}'.format(e))
            raise
    return self._wait_for_job_done(project_id, job_id)
