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
if not self._model_name:
            raise AirflowException("The model_name parameter could not be empty.")
        if not self._version_name:
            raise AirflowException("The version_name parameter could not be empty.")
    def execute(self, context):
        hook = MLEngineHook(
            gcp_conn_id=self._gcp_conn_id,
            delegate_to=self._delegate_to,
            impersonation_chain=self._impersonation_chain,
        )
        return hook.delete_version(
            project_id=self._project_id, model_name=self._model_name, version_name=self._version_name
        )
class AIPlatformConsoleLink(BaseOperatorLink):
    """
    Helper class for constructing AI Platform Console link.
    """
    name = "AI Platform Console"
    def get_link(self, operator, dttm):
        task_instance = TaskInstance(task=operator, execution_date=dttm)
        gcp_metadata_dict = task_instance.xcom_pull(task_ids=operator.task_id, key="gcp_metadata")
        if not gcp_metadata_dict:
            return ''
        job_id = gcp_metadata_dict['job_id']
        project_id = gcp_metadata_dict['project_id']
        console_link = f"https://console.cloud.google.com/ai-platform/jobs/{job_id}?project={project_id}"
        return console_link
class MLEngineStartTrainingJobOperator(BaseOperator):