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


tests/providers/google/cloud/operators/test_mlengine.py
prediction_task = MLEngineStartBatchPredictionJobOperator(
            job_id='test_prediction',
            project_id='test-project',
            region=input_with_model['region'],
            data_format=input_with_model['dataFormat'],
            input_paths=input_with_model['inputPaths'],
            output_path=input_with_model['outputPath'],
            model_name=input_with_model['modelName'].split('/')[-1],
            labels={'some': 'labels'},
            dag=self.dag,
            task_id='test-prediction',
        )
        prediction_output = prediction_task.execute(None)
        mock_hook.assert_called_once_with(
            'google_cloud_default',
            None,
            impersonation_chain=None,
        )
        hook_instance.create_job.assert_called_once_with(
            project_id='test-project',
            job={
                'jobId': 'test_prediction',
                'labels': {'some': 'labels'},
                'predictionInput': input_with_model,
            },
            use_existing_job_fn=ANY,
        )
        self.assertEqual(success_message['predictionOutput'], prediction_output)
    @patch('airflow.providers.google.cloud.operators.mlengine.MLEngineHook')
    def test_success_with_version(self, mock_hook):