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
prediction_request['predictionInput']['runtimeVersion'] = self._runtime_version
        if self._signature_name:
            prediction_request['predictionInput']['signatureName'] = self._signature_name
        hook = MLEngineHook(
            self._gcp_conn_id, self._delegate_to, impersonation_chain=self._impersonation_chain
        )
        
        
        def check_existing_job(existing_job):
            return existing_job.get('predictionInput') == prediction_request['predictionInput']
        finished_prediction_job = hook.create_job(
            project_id=self._project_id, job=prediction_request, use_existing_job_fn=check_existing_job
        )
        if finished_prediction_job['state'] != 'SUCCEEDED':
            self.log.error('MLEngine batch prediction job failed: %s', str(finished_prediction_job))
            raise RuntimeError(finished_prediction_job['errorMessage'])
        return finished_prediction_job['predictionOutput']
class MLEngineManageModelOperator(BaseOperator):
    """
    Operator for managing a Google Cloud ML Engine model.
    .. warning::
       This operator is deprecated. Consider using operators for specific operations:
       MLEngineCreateModelOperator, MLEngineGetModelOperator.
    :param model: A dictionary containing the information about the model.
        If the `operation` is `create`, then the `model` parameter should
        contain all the information about this model such as `name`.
        If the `operation` is `get`, the `model` parameter
        should contain the `name` of the model.
    :type model: dict
    :param operation: The operation to perform. Available operations are: