# -*- coding: utf-8 -*- # Copyright 2022 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import copy import dataclasses import json import re import time from typing import Any, Dict, List, Optional from google.auth import credentials as auth_credentials from google.cloud.aiplatform import base, initializer, utils from google.cloud.aiplatform.compat.types import ( explanation_v1beta1 as explanation, job_state_v1beta1 as gca_job_state, model_monitor_v1beta1 as gca_model_monitor_compat, model_monitoring_alert_v1beta1 as model_monitoring_alert, model_monitoring_job_v1beta1 as gca_model_monitoring_job_compat, model_monitoring_service_v1beta1 as model_monitoring_service, model_monitoring_spec_v1beta1 as model_monitoring_spec, model_monitoring_stats_v1beta1 as model_monitoring_stats, schedule_service_v1beta1 as gca_schedule_service, schedule_v1beta1 as gca_schedule, ) from vertexai.resources.preview.ml_monitoring.spec import ( notification, objective, output, schema, ) import proto from google.protobuf import field_mask_pb2 from google.protobuf import timestamp_pb2 from google.type import interval_pb2 from google.protobuf import text_format try: import tensorflow as tf except ImportError: tf = None try: import tensorflow_data_validation as tfdv except ImportError: tfdv = None try: from tensorflow_metadata.proto.v0 import statistics_pb2 from tensorflow_metadata.proto.v0 import anomalies_pb2 except ImportError: statistics_pb2 = None anomalies_pb2 = None _LOGGER = base.Logger(__name__) _JOB_COMPLETE_STATES = ( gca_job_state.JobState.JOB_STATE_SUCCEEDED, gca_job_state.JobState.JOB_STATE_FAILED, gca_job_state.JobState.JOB_STATE_PARTIALLY_SUCCEEDED, ) _JOB_ERROR_STATES = (gca_job_state.JobState.JOB_STATE_FAILED,) # _block_until_complete wait times _JOB_WAIT_TIME = 5 # start at five seconds _LOG_WAIT_TIME = 5 _MAX_WAIT_TIME = 60 * 5 # 5 minute wait _WAIT_TIME_MULTIPLIER = 2 # scale wait by 2 every iteration def _visualize_stats(baseline_stats_output: str, target_stats_output: str) -> None: """Visualizes the model monitoring stats from output directory.""" import tensorflow as tf if not statistics_pb2: raise TypeError("statistics_pb2 should be installed to visualize the results") if not tf.io.gfile.exists(target_stats_output): raise ValueError("No stats were generated.") if tf.io.gfile.exists(baseline_stats_output): with tf.io.gfile.GFile( baseline_stats_output, "rb" ) as baseline, tf.io.gfile.GFile(target_stats_output, "rb") as target: baseline_combined_stats = statistics_pb2.DatasetFeatureStatisticsList() baseline_combined_stats.ParseFromString(baseline.read()) target_combined_stats = statistics_pb2.DatasetFeatureStatisticsList() target_combined_stats.ParseFromString(target.read()) baseline.close() target.close() tfdv.visualize_statistics( lhs_statistics=baseline_combined_stats, rhs_statistics=target_combined_stats, lhs_name="Baseline Stats", rhs_name="Target Stats", ) else: with tf.io.gfile.GFile(target_stats_output, "rb") as target: target_combined_stats = statistics_pb2.DatasetFeatureStatisticsList() target_combined_stats.ParseFromString(target.read()) target.close() tfdv.visualize_statistics(target_combined_stats) def _visualize_anomalies(anomalies_output: str) -> None: """Visualizes the model monitoring anomalies from output directory.""" import tensorflow as tf if not anomalies_pb2: raise TypeError("anomalies_pb2 should be installed to visualize the results") with tf.io.gfile.GFile(anomalies_output, "r") as f: anomalies = anomalies_pb2.Anomalies() text_format.Merge(f.read(), anomalies) f.close() tfdv.display_anomalies(anomalies) def _visualize_feature_attribution(feature_attribution_output: str) -> None: """Visualizes the model monitoring feature attribution from output directory.""" import tensorflow as tf with tf.io.gfile.GFile(feature_attribution_output, "r") as f: print(json.dumps(json.loads(f.read()), indent=4)) def _feature_drift_stats_output_path(output_directory: str, job_id: str) -> (str, str): """Returns the baseline and target output paths for the model monitoring feature drift stats.""" return ( f"{output_directory}/tabular/jobs/{job_id}/feature_drift/baseline/statistics", f"{output_directory}/tabular/jobs/{job_id}/feature_drift/target/statistics", ) def _feature_drift_anomalies_output_path(output_directory: str, job_id: str) -> str: """Returns the output path for the model monitoring anomalies.""" return f"{output_directory}/tabular/jobs/{job_id}/feature_drift/anomalies.textproto" def _prediction_output_stats_output_path( output_directory: str, job_id: str ) -> (str, str): """Returns the baseline and target output paths for the model monitoring prediction output stats.""" return ( f"{output_directory}/tabular/jobs/{job_id}/output_drift/baseline/statistics", f"{output_directory}/tabular/jobs/{job_id}/output_drift/target/statistics", ) def _prediction_output_anomalies_output_path(output_directory: str, job_id: str) -> str: """Returns the output path for the model monitoring anomalies.""" return f"{output_directory}/tabular/jobs/{job_id}/output_drift/anomalies.textproto" def _feature_attribution_target_stats_output_path( output_directory: str, job_id: str ) -> str: """Returns the output path for the model monitoring stats.""" return f"{output_directory}/tabular/jobs/{job_id}/xai/target/feature_score.json" def _feature_attribution_baseline_stats_output_path( output_directory: str, job_id: str ) -> str: """Returns the output path for the model monitoring anomalies.""" return f"{output_directory}/tabular/jobs/{job_id}/xai/baseline/feature_score.json" def _transform_schema_pandas( dataset: Dict[str, str], feature_fields: Optional[List[str]] = None, ground_truth_fields: Optional[List[str]] = None, prediction_fields: Optional[List[str]] = None, ) -> schema.ModelMonitoringSchema: """Transforms the pandas schema to model monitoring schema.""" ground_truth_fields_list = list() prediction_fields_list = list() feature_fields_list = list() pandas_integer_types = ["integer", "Int32", "Int64", "UInt32", "UInt64"] pandas_string_types = [ "string", "bytes", "date", "time", "datetime64", "datetime", "mixed-integer", "inteval", "Interval", ] pandas_float_types = [ "floating", "decimal", "mixed-integer-float", "Float32", "Float64", ] for field in dataset: infer_type = dataset[field] if infer_type in pandas_string_types: data_type = "string" elif infer_type in pandas_integer_types: data_type = "integer" elif infer_type in pandas_float_types: data_type = "float" elif infer_type == "boolean": data_type = "boolean" elif infer_type == "categorical" or infer_type == "category": data_type = "categorical" else: raise ValueError(f"Unsupported data type: {infer_type}") if ground_truth_fields and field in ground_truth_fields: ground_truth_fields_list.append( schema.FieldSchema(name=field, data_type=data_type, repeated=False) ) elif prediction_fields and field in prediction_fields: prediction_fields_list.append( schema.FieldSchema(name=field, data_type=data_type, repeated=False) ) elif (feature_fields and field in feature_fields) or not feature_fields: feature_fields_list.append( schema.FieldSchema(name=field, data_type=data_type, repeated=False) ) return schema.ModelMonitoringSchema( ground_truth_fields=ground_truth_fields_list if ground_truth_fields else None, prediction_fields=prediction_fields_list if prediction_fields else None, feature_fields=feature_fields_list, ) def _transform_field_schema( field_schema: gca_model_monitor_compat.ModelMonitoringSchema.FieldSchema, ) -> Dict[str, Any]: result = dict() result["name"] = field_schema.name result["data_type"] = field_schema.data_type result["repeated"] = field_schema.repeated return result def _get_schedule_name(schedule_name: str) -> str: if schedule_name: client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, ) if client.parse_schedule_path(schedule_name): return schedule_name elif re.match("^{}$".format("[0-9]{0,127}"), schedule_name): return client.schedule_path( project=initializer.global_config.project, location=initializer.global_config.location, schedule=schedule_name, ) else: raise ValueError( "schedule name must be of the format `projects/{project}/locations/{location}/schedules/{schedule}` or `{schedule}`" ) return schedule_name def _get_model_monitoring_job_name( model_monitoring_job_name: str, model_monitor_name: str, ) -> str: if model_monitoring_job_name: client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, ) if client.parse_model_monitoring_job_path(model_monitoring_job_name): return model_monitoring_job_name elif re.match("^{}$".format("[0-9]{0,127}"), model_monitoring_job_name): model_monitor_name = model_monitor_name.split("/")[-1] return client.model_monitoring_job_path( project=initializer.global_config.project, location=initializer.global_config.location, model_monitor=model_monitor_name, model_monitoring_job=model_monitoring_job_name, ) else: raise ValueError( "model monitoring job name must be of the format `projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}` or `{model_monitoring_job}`" ) return model_monitoring_job_name @dataclasses.dataclass class MetricsSearchResponse: """MetricsSearchResponse represents a response of the search metrics request. Attributes: monitoring_stats (List[model_monitoring_stats.ModelMonitoringStats]): Stats retrieved for requested objectives. next_page_token (str): The page token that can be used by the next call. """ next_page_token: str _search_metrics_response: Any monitoring_stats: List[model_monitoring_stats.ModelMonitoringStats] = ( dataclasses.field(default_factory=list) ) @property def raw_search_metrics_response( self, ) -> model_monitoring_service.SearchModelMonitoringStatsResponse: """Raw search metrics response.""" return self._search_metrics_response # TODO: b/307946658 - Return a dict or a new dataclass for search_alert @dataclasses.dataclass class AlertsSearchResponse: """AlertsSearchResponse represents a response of the search alerts request. Attributes: next_page_token (str): The page token that can be used by the next call. model_monitoring_alerts (List[model_monitoring_alert.ModelMonitoringAlert]): Alerts retrieved for requested objectives. total_alerts (int): Total number of alerts retrieved for requested objectives. """ next_page_token: str _search_alerts_response: Any total_alerts: int model_monitoring_alerts: List[model_monitoring_alert.ModelMonitoringAlert] = ( dataclasses.field(default_factory=list) ) @property def raw_search_alerts_response( self, ) -> model_monitoring_service.SearchModelMonitoringAlertsResponse: """Raw search metrics response.""" return self._search_alerts_response @dataclasses.dataclass class ListJobsResponse: """ListJobsResponse represents a response of the list jobs request. Attributes: list_jobs (List[model_monitoring_job.ModelMonitoringJob]): Jobs retrieved for request. next_page_token (str): The page token that can be used by the next call. """ next_page_token: str _list_jobs_response: Any list_jobs: List[gca_model_monitoring_job_compat.ModelMonitoringJob] = ( dataclasses.field(default_factory=list) ) @property def raw_list_jobs_response( self, ) -> model_monitoring_service.ListModelMonitoringJobsResponse: """Raw list jobs response.""" return self._list_jobs_response @dataclasses.dataclass class ListSchedulesResponse: """ListSchedulesResponse represents a response of the list jobs request. Attributes: list_schedules (List[schedule.Schedule]): Jobs retrieved for request. next_page_token (str): The page token that can be used by the next call. """ next_page_token: str _list_schedules_response: Any list_schedules: List[gca_schedule.Schedule] = dataclasses.field( default_factory=list ) @property def raw_list_schedules_response(self) -> gca_schedule_service.ListSchedulesResponse: """Raw list jobs response.""" return self._list_schedules_response class ModelMonitor(base.VertexAiResourceNounWithFutureManager): """Initializer for ModelMonitor. Args: model_monitor_name (str): Required. A fully-qualified model monitor resource name or model monitor ID. Example: "projects/123/locations/us-central1/modelMonitors/456" or "456" when project and location are initialized or passed. project (str): Required. Project to retrieve model monitor from. If not set, project set in aiplatform.init will be used. location (str): Required. Location to retrieve model monitor from. If not set, location set in aiplatform.init will be used. credentials (auth_credentials.Credentials): Optional. Custom credentials to use to retrieve this model monitor. Overrides credentials set in aiplatform.init. """ client_class = utils.ModelMonitoringClientWithOverride _resource_noun = "modelMonitors" _getter_method = "get_model_monitor" _list_method = "list_model_monitors" _delete_method = "delete_model_monitor" _parse_resource_name_method = "parse_model_monitor_path" _format_resource_name_method = "model_monitor_path" def __init__( self, model_monitor_name: str, project: Optional[str] = None, location: Optional[str] = None, credentials: Optional[auth_credentials.Credentials] = None, ): super().__init__( project=project, location=location, credentials=credentials, resource_name=model_monitor_name, ) self._gca_resource = self._get_gca_resource(resource_name=model_monitor_name) @classmethod def create( cls, model_name: str, model_version_id: str, training_dataset: Optional[objective.MonitoringInput] = None, display_name: Optional[str] = None, model_monitoring_schema: Optional[schema.ModelMonitoringSchema] = None, tabular_objective_spec: Optional[objective.TabularObjective] = None, output_spec: Optional[output.OutputSpec] = None, notification_spec: Optional[notification.NotificationSpec] = None, explanation_spec: Optional[explanation.ExplanationSpec] = None, project: Optional[str] = None, location: Optional[str] = None, credentials: Optional[auth_credentials.Credentials] = None, model_monitor_id: Optional[str] = None, ) -> "ModelMonitor": """Creates a new ModelMonitor. Args: model_name (str): Required. A model resource name as model monitoring target. Format: ``projects/{project}/locations/{location}/models/{model}`` model_version_id (str): Required. Model version id. training_dataset (objective.MonitoringInput): Optional. Training dataset used to train the model. It can serve as a baseline dataset to identify changes in production. display_name (str): Optional. The user-defined name of the ModelMonitor. The name can be up to 128 characters long and can comprise any UTF-8 character. Display name of the ModelMonitor. model_monitoring_schema (schema.ModelMonitoringSchema): Required for most models, but optional for Vertex AI AutoML Tables unless the schema information is not available. The Monitoring Schema specifies the model's features, prediction outputs and ground truth properties. It is used to extract pertinent data from the dataset and to process features based on their properties. Make sure that the schema aligns with your dataset, if it does not, Vertex AI will be unable to extract data form the dataset. tabular_objective_spec (objective.TabularObjective): Optional. The default tabular monitoring objective spec for the model monitor. It can be overriden in the ModelMonitoringJob objective spec. output_spec (output.OutputSpec): Optional. The default monitoring metrics/logs export spec, it can be overriden in the ModelMonitoringJob output spec. If not specified, a default Google Cloud Storage bucket will be created under your project. notification_spec (notification.NotificationSpec): Optional. The default notification spec for monitoring result. It can be overriden in the ModelMonitoringJob notification spec. explanation_spec (explanation.ExplanationSpec): Optional. The default explanation spec for feature attribution monitoring. It can be overriden in the ModelMonitoringJob explanation spec. project (str): Optional. Project to retrieve model monitor from. If not set, project set in aiplatform.init will be used. location (str): Optional. Location to retrieve model monitor from. If not set, location set in aiplatform.init will be used. credentials (auth_credentials.Credentials): Optional. Custom credentials to use to create this model monitor. Overrides credentials set in aiplatform.init. model_monitor_id (str): Optional. The unique ID of the model monitor, which will become the final component of the model monitor resource name. If not specified, it will be generated by Vertex AI. Returns: ModelMonitor: The model monitor that was created. """ api_client = initializer.global_config.create_client( client_class=cls.client_class, credentials=credentials, location_override=location, ) if display_name: utils.validate_display_name(display_name) else: display_name = cls._generate_display_name() project = project or initializer.global_config.project location = location or initializer.global_config.location user_monitoring_target = gca_model_monitor_compat.ModelMonitor.ModelMonitoringTarget( vertex_model=gca_model_monitor_compat.ModelMonitor.ModelMonitoringTarget.VertexModelSource( model=model_name, model_version_id=model_version_id ) ) operation_future = api_client.create_model_monitor( request=model_monitoring_service.CreateModelMonitorRequest( parent=initializer.global_config.common_location_path( project=project, location=location ), model_monitor=gca_model_monitor_compat.ModelMonitor( display_name=display_name, model_monitoring_target=user_monitoring_target, training_dataset=( training_dataset._as_proto() if training_dataset else None ), model_monitoring_schema=( model_monitoring_schema._as_proto() if model_monitoring_schema else None ), tabular_objective=( tabular_objective_spec._as_proto() if tabular_objective_spec else None ), notification_spec=( notification_spec._as_proto() if notification_spec else None ), output_spec=output_spec._as_proto() if output_spec else None, explanation_spec=explanation_spec, ), model_monitor_id=model_monitor_id, ), ) _LOGGER.log_create_with_lro(cls, operation_future) created_model_monitor = operation_future.result(timeout=None) _LOGGER.log_create_complete(cls, created_model_monitor, "model_monitor") self = cls._construct_sdk_resource_from_gapic( gapic_resource=created_model_monitor, project=project, location=location, credentials=credentials, ) model_monitor_id = self._gca_resource.name.split("/")[-1] _LOGGER.info( f"https://console.cloud.google.com/vertex-ai/model-monitoring/locations/{location}/model-monitors/{model_monitor_id}?project={project}" ) return self def update( self, display_name: Optional[str] = None, training_dataset: Optional[objective.MonitoringInput] = None, model_monitoring_schema: Optional[schema.ModelMonitoringSchema] = None, tabular_objective_spec: Optional[objective.TabularObjective] = None, output_spec: Optional[output.OutputSpec] = None, notification_spec: Optional[notification.NotificationSpec] = None, explanation_spec: Optional[explanation.ExplanationSpec] = None, ) -> "ModelMonitor": """Updates an existing ModelMonitor. Args: display_name (str): Optional. The user-defined name of the ModelMonitor. The name can be up to 128 characters long and can comprise any UTF-8 character. Display name of the ModelMonitor. training_dataset (objective.MonitoringInput): Optional. Training dataset used to train the model. It can serve as a baseline dataset to identify changes in production. model_monitoring_schema (schema.ModelMonitoringSchema): Optional. The Monitoring Schema specifies the model's features, prediction outputs and ground truth properties. It is used to extract pertinent data from the dataset and to process features based on their properties. Make sure that the schema aligns with your dataset, if it does not, Vertex AI will be unable to extract data form the dataset. tabular_objective_spec (objective.TabularObjective): Optional. The default tabular monitoring objective spec for the model monitor. It can be overriden in the ModelMonitoringJob objective spec. output_spec (output.OutputSpec): Optional. The default monitoring metrics/logs export spec, it can be overriden in the ModelMonitoringJob output spec. notification_spec (notification.NotificationSpec): Optional. The default notification spec for monitoring result. It can be overriden in the ModelMonitoringJob notification spec. explanation_spec (explanation.ExplanationSpec): Optional. The default explanation spec for feature attribution monitoring. It can be overriden in the ModelMonitoringJob explanation spec. Returns: ModelMonitor: The updated model monitor. """ self._sync_gca_resource() current_monitor = copy.deepcopy(self._gca_resource) update_mask: List[str] = [] if display_name is not None: update_mask.append("display_name") current_monitor.display_name = display_name if training_dataset is not None: update_mask.append("training_dataset") current_monitor.training_dataset = training_dataset._as_proto() if model_monitoring_schema is not None: update_mask.append("model_monitoring_schema") current_monitor.model_monitoring_schema = ( model_monitoring_schema._as_proto() ) if tabular_objective_spec is not None: update_mask.append("tabular_objective") current_monitor.tabular_objective = tabular_objective_spec._as_proto() if output_spec is not None: update_mask.append("output_spec") current_monitor.output_spec = output_spec._as_proto() if notification_spec is not None: update_mask.append("notification_spec") current_monitor.notification_spec = notification_spec._as_proto() if explanation_spec is not None: update_mask.append("explanation_spec") current_monitor.explanation_spec = explanation_spec lro = self.api_client.update_model_monitor( model_monitor=current_monitor, update_mask=field_mask_pb2.FieldMask(paths=update_mask), ) self._gca_resource = lro.result() return self @base.optional_sync() def delete(self, force: bool = False, sync: bool = True) -> None: """Force delete the model monitor. Args: force (bool): Required. If force is set to True, all schedules on this ModelMonitor will be deleted first. Default is False. sync (bool): Whether to execute this method synchronously. If False, this method will be executed in concurrent Future and any downstream object will be immediately returned and synced when the Future has completed. Default is True. """ _LOGGER.log_action_start_against_resource("Deleting", "", self) lro = self.api_client.delete_model_monitor( request=model_monitoring_service.DeleteModelMonitorRequest( name=self._gca_resource.name, force=force ) ) _LOGGER.log_action_started_against_resource_with_lro( "Delete", "", self.__class__, lro ) _LOGGER.log_action_completed_against_resource("deleted.", "", self) def create_schedule( self, cron: str, target_dataset: objective.MonitoringInput, display_name: Optional[str] = None, model_monitoring_job_display_name: Optional[str] = None, start_time: Optional[timestamp_pb2.Timestamp] = None, end_time: Optional[timestamp_pb2.Timestamp] = None, tabular_objective_spec: Optional[objective.TabularObjective] = None, baseline_dataset: Optional[objective.MonitoringInput] = None, output_spec: Optional[output.OutputSpec] = None, notification_spec: Optional[notification.NotificationSpec] = None, explanation_spec: Optional[explanation.ExplanationSpec] = None, ) -> "gca_schedule.Schedule": """Creates a new Scheduled run for model monitoring job. Args: cron (str): Required. Cron schedule (https://en.wikipedia.org/wiki/Cron) to launch scheduled runs. To explicitly set a timezone to the cron tab, apply a prefix in the cron tab: "CRON_TZ=${IANA_TIME_ZONE}" or "TZ=${IANA_TIME_ZONE}". The ${IANA_TIME_ZONE} may only be a valid string from IANA time zone database. For example, "CRON_TZ=America/New_York 1 * * * *", or "TZ=America/New_York 1 * * * *". target_dataset (objective.MonitoringInput): Required. The target dataset for analysis. display_name (str): Optional. The user-defined name of the Schedule. The name can be up to 128 characters long and can be consist of any UTF-8 characters. Display name of the Schedule. model_monitoring_job_display_name (str): Optional. The user-defined name of the ModelMonitoringJob. The name can be up to 128 characters long and can be consist of any UTF-8 characters. Display name of the ModelMonitoringJob. start_time (timestamp_pb2.Timestamp): Optional. Timestamp after which the first run can be scheduled. Default to Schedule create time if not specified. end_time (timestamp_pb2.Timestamp): Optional. Timestamp after which no new runs can be scheduled. If specified, The schedule will be completed when the end_time is reached. If not specified, new runs will keep getting scheduled until this Schedule is paused or deleted. Already scheduled runs will be allowed to complete. Unset if not specified. tabular_objective_spec (objective.TabularObjective): Optional. The tabular monitoring objective spec. If not set, the default tabular objective spec in ModelMonitor will be used. You must either set here or set the default one in the ModelMonitor. baseline_dataset (objective.MonitoringInput): Optional. The baseline dataset for monitoring job. If not set, the training dataset in ModelMonitor will be used as baseline dataset. output_spec (output.OutputSpec): Optional. The monitoring metrics/logs export spec. If not set, will use the default output_spec defined in ModelMonitor. notification_spec (notification.NotificationSpec): Optional. The notification spec for monitoring result. If not set, will use the default notification_spec defined in ModelMonitor. explanation_spec (explanation.ExplanationSpec): Optional. The explanation spec for feature attribution monitoring. If not set, will use the default explanation_spec defined in ModelMonitor. Returns: Schedule: The created schedule. """ api_client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, credentials=self.credentials, location_override=self.location, ) model_monitor_name = utils.full_resource_name( resource_name=self._gca_resource.name, resource_noun=self._resource_noun, parse_resource_name_method=self._parse_resource_name, format_resource_name_method=self._format_resource_name, project=self.project, location=self.location, ) schedule_request = gca_schedule_service.CreateScheduleRequest( parent=initializer.global_config.common_location_path( project=self.project, location=self.location ), schedule=gca_schedule.Schedule( display_name=display_name, start_time=start_time, end_time=end_time, cron=cron, create_model_monitoring_job_request=model_monitoring_service.CreateModelMonitoringJobRequest( parent=model_monitor_name, model_monitoring_job=gca_model_monitoring_job_compat.ModelMonitoringJob( display_name=model_monitoring_job_display_name, model_monitoring_spec=model_monitoring_spec.ModelMonitoringSpec( objective_spec=model_monitoring_spec.ModelMonitoringObjectiveSpec( tabular_objective=( tabular_objective_spec._as_proto() if tabular_objective_spec else self._gca_resource.tabular_objective ), target_dataset=target_dataset._as_proto(), baseline_dataset=( baseline_dataset._as_proto() if baseline_dataset else self._gca_resource.training_dataset ), explanation_spec=( explanation_spec if explanation_spec else self._gca_resource.explanation_spec ), ), output_spec=( output_spec._as_proto() if output_spec else self._gca_resource.output_spec ), notification_spec=( notification_spec._as_proto() if notification_spec else self._gca_resource.notification_spec ), ), ), ), max_concurrent_run_count=1, ), ) created_schedule = api_client.select_version("v1beta1").create_schedule( request=schedule_request ) _LOGGER.log_create_complete(gca_schedule.Schedule, created_schedule, "schedule") return created_schedule def update_schedule( self, schedule_name: str, display_name: Optional[str] = None, model_monitoring_job_display_name: Optional[str] = None, cron: Optional[str] = None, baseline_dataset: Optional[objective.MonitoringInput] = None, target_dataset: Optional[objective.MonitoringInput] = None, tabular_objective_spec: Optional[objective.TabularObjective] = None, output_spec: Optional[output.OutputSpec] = None, notification_spec: Optional[notification.NotificationSpec] = None, explanation_spec: Optional[explanation.ExplanationSpec] = None, end_time: Optional[timestamp_pb2.Timestamp] = None, ) -> "gca_schedule.Schedule": """Updates an existing Schedule. Args: schedule_name (str): Required. The resource name of schedule that needs to be updated. Format: ``projects/{project}/locations/{location}/schedules/{schedule}`` or ``{schedule}`` display_name (str): Optional. The user-defined name of the Schedule. The name can be up to 128 characters long and can be consist of any UTF-8 characters. Display name of the Schedule. model_monitoring_job_display_name (str): Optional. The user-defined display name of the ModelMonitoringJob that needs to be updated. cron (str): Optional. Cron schedule (https://en.wikipedia.org/wiki/Cron) to launch scheduled runs. To explicitly set a timezone to the cron tab, apply a prefix in the cron tab: "CRON_TZ=${IANA_TIME_ZONE}" or "TZ=${IANA_TIME_ZONE}". The ${IANA_TIME_ZONE} may only be a valid string from IANA time zone database. For example, "CRON_TZ=America/New_York 1 * * * *", or "TZ=America/New_York 1 * * * *". baseline_dataset (objective.MonitoringInput): Optional. The baseline dataset for monitoring job. target_dataset (objective.MonitoringInput): Optional. The target dataset for analysis. tabular_objective_spec (objective.TabularObjective): Optional. The tabular monitoring objective spec. output_spec (output.OutputSpec): Optional. The monitoring metrics/logs export spec. notification_spec (notification.NotificationSpec): Optional. The notification spec for monitoring result. explanation_spec (explanation.ExplanationSpec): Optional. The explanation spec for feature attribution monitoring. end_time (timestamp_pb2.Timestamp): Optional. Timestamp after which no new runs can be scheduled. Returns: Schedule: The updated schedule. """ api_client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, credentials=self.credentials, location_override=self.location, ) model_monitor_name = utils.full_resource_name( resource_name=self._gca_resource.name, resource_noun=self._resource_noun, parse_resource_name_method=self._parse_resource_name, format_resource_name_method=self._format_resource_name, project=self.project, location=self.location, ) schedule_name = _get_schedule_name(schedule_name) current_schedule = copy.deepcopy(self.get_schedule(schedule_name=schedule_name)) update_mask = [] if display_name is not None: update_mask.append("display_name") current_schedule.display_name = display_name if cron is not None: update_mask.append("cron") current_schedule.cron = cron if end_time is not None: update_mask.append("end_time") current_schedule.end_time = end_time current_job_request = current_schedule.create_model_monitoring_job_request current_spec = current_job_request.model_monitoring_job.model_monitoring_spec if ( tabular_objective_spec is not None or output_spec is not None or notification_spec is not None or model_monitoring_job_display_name is not None or baseline_dataset is not None or target_dataset is not None ): update_mask.append("create_model_monitoring_job_request") updated_model_monitoring_spec = model_monitoring_spec.ModelMonitoringSpec( objective_spec=model_monitoring_spec.ModelMonitoringObjectiveSpec( tabular_objective=( tabular_objective_spec._as_proto() if tabular_objective_spec else current_spec.objective_spec.tabular_objective ), baseline_dataset=( baseline_dataset._as_proto() if baseline_dataset else current_spec.objective_spec.baseline ), target_dataset=( target_dataset._as_proto() if target_dataset else current_spec.objective_spec.target ), explanation_spec=( explanation_spec if explanation_spec else current_spec.objective_spec.explanation_spec ), ), output_spec=( output_spec._as_proto() if output_spec else current_spec.output_spec ), notification_spec=( notification_spec._as_proto() if notification_spec else current_spec.notification_spec ), ) current_schedule.create_model_monitoring_job_request = model_monitoring_service.CreateModelMonitoringJobRequest( parent=model_monitor_name, model_monitoring_job=gca_model_monitoring_job_compat.ModelMonitoringJob( display_name=( model_monitoring_job_display_name if model_monitoring_job_display_name else current_job_request.model_monitoring_job.display_name ), model_monitoring_spec=updated_model_monitoring_spec, ), ) return api_client.select_version("v1beta1").update_schedule( schedule=current_schedule, update_mask=field_mask_pb2.FieldMask(paths=update_mask), ) def delete_schedule(self, schedule_name: str) -> None: """Deletes an existing Schedule. Args: schedule_name (str): Required. The resource name of schedule that needs to be deleted. Format: ``projects/{project}/locations/{location}/schedules/{schedule}`` or ``{schedule}`` """ api_client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, credentials=self.credentials, location_override=self.location, ) schedule_name = _get_schedule_name(schedule_name) return api_client.select_version("v1beta1").delete_schedule(name=schedule_name) def pause_schedule(self, schedule_name: str) -> None: """Pauses an existing Schedule. Args: schedule_name (str): Required. The resource name of schedule that needs to be paused. Format: ``projects/{project}/locations/{location}/schedules/{schedule}`` or ``{schedule}`` """ api_client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, credentials=self.credentials, location_override=self.location, ) schedule_name = _get_schedule_name(schedule_name) return api_client.select_version("v1beta1").pause_schedule(name=schedule_name) def resume_schedule(self, schedule_name: str) -> None: """Resumes an existing Schedule. Args: schedule_name (str): Required. The resource name of schedule that needs to be resumed. Format: ``projects/{project}/locations/{location}/schedules/{schedule}`` or ``{schedule}`` """ api_client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, credentials=self.credentials, location_override=self.location, ) schedule_name = _get_schedule_name(schedule_name) return api_client.select_version("v1beta1").resume_schedule(name=schedule_name) def get_schedule(self, schedule_name: str) -> "gca_schedule.Schedule": """Gets an existing Schedule. Args: schedule_name (str): Required. The resource name of schedule that needs to be fetched. Format: ``projects/{project}/locations/{location}/schedules/{schedule}`` or ``{schedule}`` Returns: Schedule: The schedule requested. """ api_client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, credentials=self.credentials, location_override=self.location, ) schedule_name = _get_schedule_name(schedule_name) return api_client.select_version("v1beta1").get_schedule(name=schedule_name) def list_schedules( self, filter: Optional[str] = None, page_size: Optional[int] = None, page_token: Optional[str] = None, ) -> "ListSchedulesResponse.list_schedules": """List Schedules. Args: filter (str): Optional. Lists the Schedules that match the filter expression. The following fields are supported: - ``display_name``: Supports ``=``, ``!=`` comparisons, and ``:`` wildcard. - ``state``: Supports ``=`` and ``!=`` comparisons. - ``request``: Supports existence of the check. (e.g. ``create_pipeline_job_request:*`` --> Schedule has create_pipeline_job_request). - ``create_time``: Supports ``=``, ``!=``, ``<``, ``>``, ``<=``, and ``>=`` comparisons. Values must be in RFC 3339 format. - ``start_time``: Supports ``=``, ``!=``, ``<``, ``>``, ``<=``, and ``>=`` comparisons. Values must be in RFC 3339 format. - ``end_time``: Supports ``=``, ``!=``, ``<``, ``>``, ``<=``, ``>=`` comparisons and ``:*`` existence check. Values must be in RFC 3339 format. - ``next_run_time``: Supports ``=``, ``!=``, ``<``, ``>``, ``<=``, and ``>=`` comparisons. Values must be in RFC 3339 format. Filter expressions can be combined together using logical operators (``NOT``, ``AND`` & ``OR``). The syntax to define filter expression is based on https://google.aip.dev/160. page_size (int): Optional. The standard page list size. page_token (str): Optional. A page token received from a previous call. Returns: MetricsSearchResponse: The model monitoring stats results. """ api_client = initializer.global_config.create_client( client_class=utils.ScheduleClientWithOverride, credentials=self.credentials, location_override=self.location, ) filter = ( f"{filter} AND create_model_monitoring_job_request.parent={self._gca_resource.name}" if filter else f"create_model_monitoring_job_request.parent={self._gca_resource.name}" ) list_schedules_response = ( api_client.select_version("v1beta1") .list_schedules( request=gca_schedule_service.ListSchedulesRequest( parent=f"projects/{self.project}/locations/{self.location}", filter=filter, page_size=page_size, page_token=page_token, ) ) ._response ) return ListSchedulesResponse( list_schedules=list_schedules_response.schedules, next_page_token=list_schedules_response.next_page_token, _list_schedules_response=list_schedules_response, ).list_schedules def run( self, target_dataset: objective.MonitoringInput, display_name: Optional[str] = None, model_monitoring_job_id: Optional[str] = None, sync: Optional[bool] = False, tabular_objective_spec: Optional[objective.TabularObjective] = None, baseline_dataset: Optional[objective.MonitoringInput] = None, output_spec: Optional[output.OutputSpec] = None, notification_spec: Optional[notification.NotificationSpec] = None, explanation_spec: Optional[explanation.ExplanationSpec] = None, ) -> "ModelMonitoringJob": """Creates a new ModelMonitoringJob. Args: target_dataset (objective.MonitoringInput): Required. The target dataset for analysis. display_name (str): Optional. The user-defined name of the ModelMonitoringJob. The name can be up to 128 characters long and can comprise any UTF-8 character. Display name of the ModelMonitoringJob. model_monitoring_job_id (str): Optional. The unique ID of the model monitoring job run, which will become the final component of the model monitoring job resource name. The maximum length is 63 characters, and valid characters are /^[a-z]([a-z0-9-]{0,61}[a-z0-9])?$/. If not specified, it will be generated by Vertex AI. sync (bool): Whether to execute this method synchronously. If False, this method will be executed in concurrent Future and any downstream object will be immediately returned and synced when the Future has completed. Default is False. tabular_objective_spec (objective.TabularObjective): Optional. The tabular monitoring objective spec for the model monitoring job. baseline_dataset (objective.MonitoringInput): Optional. The baseline dataset for monitoring job. If not set, the training dataset in ModelMonitor will be used as baseline dataset. output_spec (output.OutputSpec): Optional. The monitoring metrics/logs export spec. If not set, will use the default output_spec defined in ModelMonitor. notification_spec (notification.NotificationSpec): Optional. The notification spec for monitoring result. If not set, will use the default notification_spec defined in ModelMonitor. explanation_config (explanation.ExplanationSpec): Optional. The explanation spec for feature attribution monitoring. If not set, will use the default explanation_spec defined in ModelMonitor. Returns: ModelMonitoringJob: The model monitoring job that was created. """ model_monitor_name = utils.full_resource_name( resource_name=self._gca_resource.name, resource_noun=self._resource_noun, parse_resource_name_method=self._parse_resource_name, format_resource_name_method=self._format_resource_name, project=self.project, location=self.location, ) return ModelMonitoringJob.create( model_monitor_name=model_monitor_name, project=self.project, location=self.location, credentials=self.credentials, display_name=display_name, target_dataset=target_dataset, baseline_dataset=baseline_dataset, model_monitoring_job_id=model_monitoring_job_id, tabular_objective_spec=tabular_objective_spec, output_spec=output_spec, notification_spec=notification_spec, explanation_spec=explanation_spec, sync=sync, ) def search_metrics( self, stats_name: Optional[str] = None, objective_type: Optional[str] = None, model_monitoring_job_name: Optional[str] = None, schedule_name: Optional[str] = None, algorithm: Optional[str] = None, start_time: Optional[timestamp_pb2.Timestamp] = None, end_time: Optional[timestamp_pb2.Timestamp] = None, page_size: Optional[int] = None, page_token: Optional[str] = None, ) -> "MetricsSearchResponse.monitoring_stats": """Search ModelMonitoringStats. Args: stats_name (str): Optional. The stats name filter for the search, if not set, all stats will be returned. For tabular model it's the feature name. objective_type (str): Optional. One of the supported monitoring objectives: `raw-feature-drift` `prediction-output-drift` `feature-attribution` model_monitoring_job_name (str): Optional. The resource name of a particular model monitoring job that the user wants to search metrics result from. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}`` schedule_name (str): Optional. The resource name of a particular model monitoring schedule that the user wants to search metrics result from. Format: ``projects/{project}/locations/{location}/schedules/{schedule}`` algorithm (str): Optional. The algorithm type filter for the search, eg: jensen_shannon_divergence, l_infinity. start_time (timestamp_pb2.Timestamp): Optional. Inclusive start of the time interval for which results should be returned. end_time (timestamp_pb2.Timestamp): Optional. Exclusive end of the time interval for which results should be returned. page_size (int): Optional. The standard page list size. page_token (str): Optional. A page token received from a previous call. Returns: MetricsSearchResponse: The model monitoring stats results. """ api_client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, credentials=self.credentials, location_override=self.location, ) user_time_interval = ( interval_pb2.Interval(start_time=start_time, end_time=end_time) if start_time or end_time else None ) model_monitoring_stats_response = api_client.search_model_monitoring_stats( request=model_monitoring_service.SearchModelMonitoringStatsRequest( model_monitor=self._gca_resource.name, stats_filter=model_monitoring_stats.SearchModelMonitoringStatsFilter( tabular_stats_filter=model_monitoring_stats.SearchModelMonitoringStatsFilter.TabularStatsFilter( stats_name=stats_name, objective_type=objective_type, model_monitoring_job=model_monitoring_job_name, model_monitoring_schedule=schedule_name, algorithm=algorithm, ), ), time_interval=user_time_interval, page_size=page_size, page_token=page_token, ), )._response return MetricsSearchResponse( monitoring_stats=model_monitoring_stats_response.monitoring_stats, next_page_token=model_monitoring_stats_response.next_page_token, _search_metrics_response=model_monitoring_stats_response, ).monitoring_stats def search_alerts( self, stats_name: Optional[str] = None, objective_type: Optional[str] = None, model_monitoring_job_name: Optional[str] = None, start_time: Optional[timestamp_pb2.Timestamp] = None, end_time: Optional[timestamp_pb2.Timestamp] = None, page_size: Optional[int] = None, page_token: Optional[str] = None, ) -> Dict[str, Any]: """Search ModelMonitoringAlerts. Args: stats_name (str): Optional. The stats name filter for the search, if not set, all stats will be returned. For tabular models, provide the name of the feature to return alerts from. objective_type (str): Optional. Return alerts from one of the supported monitoring objectives: `raw-feature-drift` `prediction-output-drift` `feature-attribution` model_monitoring_job_name (str): Optional. The resource name of a particular model monitoring job that the user wants to search metrics result from. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}`` start_time (timestamp_pb2.Timestamp): Optional. Inclusive start of the time interval for which alerts should be returned. end_time (timestamp_pb2.Timestamp): Optional. Exclusive end of the time interval for which alerts should be returned. page_size (int): Optional. The standard page list size. page_token (str): Optional. A page token received from a previous call. Returns: AlertsSearchResponse: The model monitoring alerts results. """ api_client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, credentials=self.credentials, location_override=self.location, ) user_time_interval = ( interval_pb2.Interval(start_time=start_time, end_time=end_time) if start_time or end_time else None ) model_monitoring_alerts_response = api_client.search_model_monitoring_alerts( request=model_monitoring_service.SearchModelMonitoringAlertsRequest( model_monitor=self._gca_resource.name, stats_name=stats_name, objective_type=objective_type, model_monitoring_job=model_monitoring_job_name, alert_time_interval=user_time_interval, page_size=page_size, page_token=page_token, ), )._response alert_response = AlertsSearchResponse( model_monitoring_alerts=model_monitoring_alerts_response.model_monitoring_alerts, next_page_token=model_monitoring_alerts_response.next_page_token, total_alerts=model_monitoring_alerts_response.total_number_alerts, _search_alerts_response=model_monitoring_alerts_response, ) return { "total_number_alerts": alert_response.total_alerts, "model_monitoring_alerts": alert_response.model_monitoring_alerts, } def list_jobs( self, page_size: Optional[int] = None, page_token: Optional[str] = None, ) -> "ListJobsResponse.list_jobs": """List ModelMonitoringJobs. Args: page_size (int): Optional. The standard page list size. page_token (str): Optional. A page token received from a previous call. Returns: ListJobsResponse.list_jobs: The list model monitoring jobs responses. """ api_client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, credentials=self.credentials, location_override=self.location, ) model_monitor_name = utils.full_resource_name( resource_name=self._gca_resource.name, resource_noun=self._resource_noun, parse_resource_name_method=self._parse_resource_name, format_resource_name_method=self._format_resource_name, project=self.project, location=self.location, ) list_jobs_response = api_client.list_model_monitoring_jobs( request=model_monitoring_service.ListModelMonitoringJobsRequest( parent=model_monitor_name, page_size=page_size, page_token=page_token, ) )._response return ListJobsResponse( list_jobs=list_jobs_response.model_monitoring_jobs, next_page_token=list_jobs_response.next_page_token, _list_jobs_response=list_jobs_response, ).list_jobs def delete_model_monitoring_job(self, model_monitoring_job_name: str) -> None: """Delete a model monitoring job. Args: model_monitoring_job_name (str): Required. The resource name of the model monitoring job that needs to be deleted. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}`` or ``{model_monitoring_job}`` """ api_client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, credentials=self.credentials, location_override=self.location, ) job_resource_name = _get_model_monitoring_job_name( model_monitoring_job_name, self._gca_resource.name ) api_client.delete_model_monitoring_job(name=job_resource_name) def get_model_monitoring_job( self, model_monitoring_job_name: str ) -> "ModelMonitoringJob": """Get the specified ModelMonitoringJob. Args: model_monitoring_job_name (str): Required. The resource name of the ModelMonitoringJob that is needed. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}`` or ``{model_monitoring_job}`` Returns: ModelMonitoringJob: The model monitoring job get. """ self.wait() job_resource_name = _get_model_monitoring_job_name( model_monitoring_job_name, self._gca_resource.name ) return ModelMonitoringJob( model_monitoring_job_name=job_resource_name, project=self.project, location=self.location, credentials=self.credentials, ) def show_feature_drift_stats(self, model_monitoring_job_name: str) -> None: """The method to visualize the feature drift result from a model monitoring job as a histogram chart and a table. Args: model_monitoring_job_name (str): Required. The resource name of model monitoring job to show the drift stats from. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}`` or ``{model_monitoring_job}`` """ api_client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, credentials=self.credentials, location_override=self.location, ) if model_monitoring_job_name.startswith("projects/"): job_resource_name = model_monitoring_job_name job_id = model_monitoring_job_name.split("/")[-1] else: job_resource_name = f"{self._gca_resource.name}/modelMonitoringJobs/{model_monitoring_job_name}" job_id = model_monitoring_job_name job = api_client.get_model_monitoring_job(name=job_resource_name) output_directory = ( job.model_monitoring_spec.output_spec.gcs_base_directory.output_uri_prefix ) target_output, baseline_output = _feature_drift_stats_output_path( output_directory, job_id ) anomoaly_output = _feature_drift_anomalies_output_path(output_directory, job_id) _visualize_stats(baseline_output, target_output) _visualize_anomalies(anomoaly_output) def get_schema(self) -> gca_model_monitor_compat.ModelMonitoringSchema: """Get the schema of the model monitor.""" self._sync_gca_resource() return self._gca_resource.model_monitoring_schema def show_output_drift_stats(self, model_monitoring_job_name: str) -> None: """The method to visualize the prediction output drift result from a model monitoring job as a histogram chart and a table. Args: model_monitoring_job_name (str): Required. The resource name of model monitoring job to show the drift stats from. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}`` or ``{model_monitoring_job}`` """ api_client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, credentials=self.credentials, location_override=self.location, ) if model_monitoring_job_name.startswith("projects/"): job_resource_name = model_monitoring_job_name job_id = model_monitoring_job_name.split("/")[-1] else: job_resource_name = f"{self._gca_resource.name}/modelMonitoringJobs/{model_monitoring_job_name}" job_id = model_monitoring_job_name job = api_client.get_model_monitoring_job(name=job_resource_name) output_directory = ( job.model_monitoring_spec.output_spec.gcs_base_directory.output_uri_prefix ) target_output, baseline_output = _prediction_output_stats_output_path( output_directory, job_id ) anomoaly_output = _prediction_output_anomalies_output_path( output_directory, job_id ) _visualize_stats(baseline_output, target_output) _visualize_anomalies(anomoaly_output) def show_feature_attribution_drift_stats( self, model_monitoring_job_name: str ) -> None: """The method to visualize the feature attribution drift result from a model monitoring job as a histogram chart and a table. Args: model_monitoring_job_name (str): Required. The resource name of model monitoring job to show the feature attribution drift stats from. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}/modelMonitoringJobs/{model_monitoring_job}`` or ``{model_monitoring_job}`` """ api_client = initializer.global_config.create_client( client_class=utils.ModelMonitoringClientWithOverride, credentials=self.credentials, location_override=self.location, ) if model_monitoring_job_name.startswith("projects/"): job_resource_name = model_monitoring_job_name job_id = model_monitoring_job_name.split("/")[-1] else: job_resource_name = f"{self._gca_resource.name}/modelMonitoringJobs/{model_monitoring_job_name}" job_id = model_monitoring_job_name job = api_client.get_model_monitoring_job(name=job_resource_name) output_directory = ( job.model_monitoring_spec.output_spec.gcs_base_directory.output_uri_prefix ) target_stats_output = _feature_attribution_target_stats_output_path( output_directory, job_id ) baseline_stats_output = _feature_attribution_baseline_stats_output_path( output_directory, job_id ) _visualize_feature_attribution(baseline_stats_output) _visualize_feature_attribution(target_stats_output) class ModelMonitoringJob(base.VertexAiStatefulResource): r"""Initializer for ModelMonitoringJob. Example Usage: my_monitoring_job = aiplatform.ModelMonitoringJob( model_monitoring_job_name='projects/123/locations/us-central1/modelMonitors/\ my_model_monitor_id/modelMonitoringJobs/my_monitoring_job_id' ) or my_monitoring_job = aiplatform.aiplatform.ModelMonitoringJob( model_monitoring_job_name='my_monitoring_job_id', model_monitor_id='my_model_monitor_id', ) Args: model_monitoring_job_name (str): Required. The resource name for the Model Monitoring Job if provided alone, or the model monitoring job id if provided with model_monitor_id. model_monitor_id (str): Optional. The model monitor id depends on the way of initializing ModelMonitoringJob. project (str): Required. Project to retrieve endpoint from. If not set, project set in aiplatform.init will be used. location (str): Required. Location to retrieve endpoint from. If not set, location set in aiplatform.init will be used. credentials (auth_credentials.Credentials): Optional. Custom credentials to use to init model monitoring job. Overrides credentials set in aiplatform.init. """ client_class = utils.ModelMonitoringClientWithOverride _resource_noun = "modelMonitoringJobs" _getter_method = "get_model_monitoring_job" _list_method = "list_model_monitoring_jobs" _delete_method = "delete_model_monitoring_job" _parse_resource_name_method = "parse_model_monitoring_job_path" _format_resource_name_method = "model_monitoring_job_path" # Required by the done() method _valid_done_states = _JOB_COMPLETE_STATES def __init__( self, model_monitoring_job_name: str, model_monitor_id: Optional[str] = None, project: Optional[str] = None, location: Optional[str] = None, credentials: Optional[auth_credentials.Credentials] = None, ): super().__init__( project=project, location=location, credentials=credentials, resource_name=model_monitoring_job_name, ) self._gca_resource = self._get_gca_resource( resource_name=model_monitoring_job_name, parent_resource_name_fields=( {ModelMonitor._resource_noun: model_monitor_id} if model_monitor_id else model_monitor_id ), ) @property def state(self) -> gca_job_state.JobState: """Fetch Job again and return the current JobState. Returns: state (job_state.JobState): Enum that describes the state of a Model Monitoring Job. """ # Fetch the Job again for most up-to-date job state self._sync_gca_resource() return self._gca_resource.state @classmethod def _construct_sdk_resource_from_gapic( cls, gapic_resource: proto.Message, project: Optional[str] = None, location: Optional[str] = None, credentials: Optional[auth_credentials.Credentials] = None, ) -> "ModelMonitoringJob": """Given a GAPIC ModelMonitoringJob object, return the SDK representation. Args: gapic_resource (proto.Message): A GAPIC representation of a ModelMonitoringJob resource, usually retrieved by a get_* or in a list_* API call. project (str): Optional. Project to construct ModelMonitoringJob object from. If not set, project set in aiplatform.init will be used. location (str): Optional. Location to construct ModelMonitoringJob object from. If not set, location set in aiplatform.init will be used. credentials (auth_credentials.Credentials): Optional. Custom credentials to use to construct ModelMonitoringJob. Overrides credentials set in aiplatform.init. Returns: ModelMonitoringJob: The model monitoring job that was created. """ model_monitoring_job = super()._construct_sdk_resource_from_gapic( gapic_resource=gapic_resource, project=project, location=location, credentials=credentials, ) return model_monitoring_job def _block_until_complete(self) -> None: """Helper method to block and check on job until complete.""" # Used these numbers so failures surface fast wait = _JOB_WAIT_TIME # start at five seconds log_wait = _LOG_WAIT_TIME max_wait = _MAX_WAIT_TIME # 5 minute wait multiplier = _WAIT_TIME_MULTIPLIER # scale wait by 2 every iteration previous_time = time.time() while not self.done(): current_time = time.time() if current_time - previous_time >= log_wait: _LOGGER.info( "%s %s current state:\n%s" % ( self.__class__.__name__, self._gca_resource.name, self._gca_resource.state, ) ) log_wait = min(log_wait * multiplier, max_wait) previous_time = current_time time.sleep(wait) # Error is only populated when the job state is JOB_STATE_FAILED. if self._gca_resource.state in _JOB_ERROR_STATES: raise RuntimeError( "Job failed with:\n%s" % self._gca_resource.job_execution_detail.error ) elif ( self._gca_resource.state == gca_job_state.JobState.JOB_STATE_PARTIALLY_SUCCEEDED ): obj_status_msg = "" for ( obj, status, ) in self._gca_resource.job_execution_detail.objective_status.items(): obj_status_msg += f"{obj}: {status}\n" _LOGGER.warning("Job partially succeeded:\n%s" % obj_status_msg) else: _LOGGER.log_action_completed_against_resource("run", "completed", self) @classmethod def create( cls, model_monitor_name: str = None, target_dataset: objective.MonitoringInput = None, display_name: Optional[str] = None, model_monitoring_job_id: Optional[str] = None, project: Optional[str] = None, location: Optional[str] = None, credentials: Optional[auth_credentials.Credentials] = None, baseline_dataset: Optional[objective.MonitoringInput] = None, tabular_objective_spec: Optional[objective.TabularObjective] = None, output_spec: Optional[output.OutputSpec] = None, notification_spec: Optional[notification.NotificationSpec] = None, explanation_spec: Optional[explanation.ExplanationSpec] = None, sync: bool = False, ) -> "ModelMonitoringJob": """Creates a new ModelMonitoringJob. Args: model_monitor_name (str): Required. The parent model monitor resource name. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}`` target_dataset (objective.MonitoringInput): Required. The target dataset for analysis. display_name (str): Optional. The user-defined name of the ModelMonitoringJob. The name can be up to 128 characters long and can comprise any UTF-8 character. model_monitoring_job_id (str): Optional. The unique ID of the model monitoring job run, which will become the final component of the model monitoring job resource name. The maximum length is 63 characters, and valid characters are /^[a-z]([a-z0-9-]{0,61}[a-z0-9])?$/. If not specified, it will be generated by Vertex AI. project (str): Optional. Project to retrieve endpoint from. If not set, project set in aiplatform.init will be used. location (str): Optional. Location to retrieve endpoint from. If not set, location set in aiplatform.init will be used. credentials (auth_credentials.Credentials): Optional. Custom credentials to use to create model monitoring job. Overrides credentials set in aiplatform.init. baseline_dataset (objective.MonitoringInput): Optional. The baseline dataset for monitoring job. If not set, the training dataset in ModelMonitor will be used as baseline dataset. output_spec (output.OutputSpec): Optional. The monitoring metrics/logs export spec. If not set, will use the default output_spec defined in ModelMonitor. notification_spec (notification.NotificationSpec): Optional. The notification spec for monitoring result. If not set, will use the default notification_spec defined in ModelMonitor. explanation_spec (explanation.ExplanationSpec): Optional. The explanation spec for feature attribution monitoring. If not set, will use the default explanation_spec defined in ModelMonitor. sync (bool): Required. Whether to execute this method synchronously. If False, this method will be executed in concurrent Future and any downstream object will be immediately returned and synced when the Future has completed. Default is False. Returns: ModelMonitoringJob: The model monitoring job that was created. """ if not display_name: display_name = cls._generate_display_name() utils.validate_display_name(display_name) project = project or initializer.global_config.project location = location or initializer.global_config.location parent = utils.full_resource_name( resource_name=model_monitor_name, resource_noun=ModelMonitor._resource_noun, parse_resource_name_method=ModelMonitor._parse_resource_name, format_resource_name_method=ModelMonitor._format_resource_name, project=project, location=location, ) gca_model_monitoring_job = gca_model_monitoring_job_compat.ModelMonitoringJob( display_name=display_name, model_monitoring_spec=model_monitoring_spec.ModelMonitoringSpec( objective_spec=model_monitoring_spec.ModelMonitoringObjectiveSpec( tabular_objective=( tabular_objective_spec._as_proto() if tabular_objective_spec else None ), baseline_dataset=( baseline_dataset._as_proto() if baseline_dataset else None ), target_dataset=( target_dataset._as_proto() if target_dataset else None ), explanation_spec=explanation_spec, ), output_spec=(output_spec._as_proto() if output_spec else None), notification_spec=( notification_spec._as_proto() if notification_spec else None ), ), ) empty_model_monitoring_job = cls._empty_constructor( project=project, location=location, credentials=credentials, ) return cls._submit_job( model_monitor_name=parent, empty_model_monitoring_job=empty_model_monitoring_job, gca_model_monitoring_job=gca_model_monitoring_job, model_monitoring_job_id=model_monitoring_job_id, sync=sync, project=project, location=location, credentials=credentials, ) @classmethod @base.optional_sync(return_input_arg="empty_model_monitoring_job") def _submit_job( cls, model_monitor_name: str, empty_model_monitoring_job: "ModelMonitoringJob", gca_model_monitoring_job: gca_model_monitoring_job_compat.ModelMonitoringJob, sync: bool = False, model_monitoring_job_id: Optional[str] = None, project: Optional[str] = None, location: Optional[str] = None, credentials: Optional[auth_credentials.Credentials] = None, ) -> "ModelMonitoringJob": """Submits a new ModelMonitoringJob. Args: model_monitor_name (str): Required. The parent model monitor resource name. Format: ``projects/{project}/locations/{location}/modelMonitors/{model_monitor}`` empty_model_monitoring_job (ModelMonitoringJob): Required. ModelMonitoringJob without _gca_resource populated. gca_model_monitoring_job (gca_model_monitoring_job_compat.ModelMonitoringJob): Required. a model monitoring job proto for creating a model monitoring job on Vertex AI. sync (bool): Required. Whether to execute this method synchronously. If False, this method will be executed in concurrent Future and any downstream object will be immediately returned and synced when the Future has completed. Default is False. model_monitoring_job_id (str): Optional. The unique ID of the model monitoring job run, which will become the final component of the model monitoring job resource name. The maximum length is 63 characters, and valid characters are /^[a-z]([a-z0-9-]{0,61}[a-z0-9])?$/. If not specified, it will be generated by Vertex AI. project (str): Optional. Project to retrieve endpoint from. If not set, project set in aiplatform.init will be used. location (str): Optional. Location to retrieve endpoint from. If not set, location set in aiplatform.init will be used. credentials (auth_credentials.Credentials): Optional. Custom credentials to use to create model monitoring job. Overrides credentials set in aiplatform.init. Returns: ModelMonitoringJob: The model monitoring job that was created. """ api_client = initializer.global_config.create_client( client_class=cls.client_class, credentials=credentials, location_override=location, ) _LOGGER.log_create_with_lro(cls) created_model_monitoring_job = api_client.create_model_monitoring_job( request=model_monitoring_service.CreateModelMonitoringJobRequest( parent=model_monitor_name, model_monitoring_job=gca_model_monitoring_job, model_monitoring_job_id=model_monitoring_job_id, ), ) empty_model_monitoring_job._gca_resource = created_model_monitoring_job model_monitoring_job = cls._construct_sdk_resource_from_gapic( gapic_resource=created_model_monitoring_job, project=project, location=location, credentials=credentials, ) _LOGGER.log_create_complete( cls, created_model_monitoring_job, "model_monitoring_job" ) model_monitoring_job._block_until_complete() return model_monitoring_job def delete(self) -> None: """Deletes an Model Monitoring Job.""" self.api_client.delete_model_monitoring_job(name=self._gca_resource.name)