Training with a custom service account needs a new field named service account
trainingInput:
  serviceAccount: CUSTOM_SERVICE_ACCOUNT

https://cloud.google.com/ai-platform/training/docs/custom-service-account


airflow/providers/google/cloud/operators/mlengine.py
return
        hook = MLEngineHook(
            gcp_conn_id=self._gcp_conn_id,
            delegate_to=self._delegate_to,
            impersonation_chain=self._impersonation_chain,
        )
        
        
        def check_existing_job(existing_job):
            existing_training_input = existing_job.get('trainingInput')
            requested_training_input = training_request['trainingInput']
            if 'scaleTier' not in existing_training_input:
                existing_training_input['scaleTier'] = None
            existing_training_input['args'] = existing_training_input.get('args')
            requested_training_input["args"] = (
                requested_training_input['args'] if requested_training_input["args"] else None
            )
            return existing_training_input == requested_training_input
        finished_training_job = hook.create_job(
            project_id=self._project_id, job=training_request, use_existing_job_fn=check_existing_job
        )
        if finished_training_job['state'] != 'SUCCEEDED':
            self.log.error('MLEngine training job failed: %s', str(finished_training_job))
            raise RuntimeError(finished_training_job['errorMessage'])
        gcp_metadata = {
            "job_id": job_id,
            "project_id": self._project_id,
        }
        context['task_instance'].xcom_push("gcp_metadata", gcp_metadata)
class MLEngineTrainingCancelJobOperator(BaseOperator):