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
self._training_args = training_args
        self._region = region
        self._scale_tier = scale_tier
        self._master_type = master_type
        self._master_config = master_config
        self._runtime_version = runtime_version
        self._python_version = python_version
        self._job_dir = job_dir
        self._gcp_conn_id = gcp_conn_id
        self._delegate_to = delegate_to
        self._mode = mode
        self._labels = labels
        self._impersonation_chain = impersonation_chain
        if not self._project_id:
            raise AirflowException('Google Cloud project id is required.')
        if not self._job_id:
            raise AirflowException('An unique job id is required for Google MLEngine training ' 'job.')
        if not package_uris:
            raise AirflowException('At least one python package is required for MLEngine ' 'Training job.')
        if not training_python_module:
            raise AirflowException(
                'Python module name to run after installing required ' 'packages is required.'
            )
        if not self._region:
            raise AirflowException('Google Compute Engine region is required.')
        if self._scale_tier is not None and self._scale_tier.upper() == "CUSTOM" and not self._master_type:
            raise AirflowException('master_type must be set when scale_tier is CUSTOM')
        if self._master_config and not self._master_type:
            raise AirflowException('master_type must be set when master_config is provided')
    def execute(self, context):