def create_job(self, project_id, job, use_existing_job_fn=None):
request = self._mlengine.projects().jobs().create(<EOL>parent='<STR_LIT>'.format(project_id),<EOL>body=job)<EOL>job_id = job['<STR_LIT>']<EOL>try:<EOL><INDENT>request.execute()<EOL><DEDENT>except HttpError as e:<EOL><INDENT>if e.resp.status == <NUM_LIT>:<EOL><INDENT>if use_existing_job_fn is not None:<EOL><INDENT>existing_job = self._get_job(project_id, job_id)<EOL>if not use_existing_job_fn(existing_job):<EOL><INDENT>self.log.error(<EOL>'<STR_LIT>'<EOL>'<STR_LIT>',<EOL>job_id, existing_job<EOL>)<EOL>raise<EOL><DEDENT><DEDENT>self.log.info(<EOL>'<STR_LIT>',<EOL>job_id<EOL>)<EOL><DEDENT>else:<EOL><INDENT>self.log.error('<STR_LIT>'.format(e))<EOL>raise<EOL><DEDENT><DEDENT>return self._wait_for_job_done(project_id, job_id)<EOL>
Launches a MLEngine job and wait for it to reach a terminal state.

:param project_id: The Google Cloud project id within which MLEngine
    job will be launched.
:type project_id: str

:param job: MLEngine Job object that should be provided to the MLEngine
    API, such as: ::

        {
          'jobId': 'my_job_id',
          'trainingInput': {
            'scaleTier': 'STANDARD_1',
            ...
          }
        }

:type job: dict

:param use_existing_job_fn: In case that a MLEngine 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 MLEngine 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 MLEngine job.
:type use_existing_job_fn: function

:return: The MLEngine job object if the job successfully reach a
    terminal state (which might be FAILED or CANCELLED state).
:rtype: dict