Merge branch 'main' into ODSC-55290/update_job_byoc_api

Author: lu-ohai

ads/opctl/operator/lowcode/anomaly/const.py

@@ -94,3 +94,4 @@ class OutputColumns(str, metaclass=ExtendedEnumMeta):
 
 
 TODS_DEFAULT_MODEL = "ocsvm"
+SUBSAMPLE_THRESHOLD = 1000

ads/opctl/operator/lowcode/anomaly/model/base_model.py

@@ -16,7 +16,7 @@
 
 from ads.common.object_storage_details import ObjectStorageDetails
 from ads.opctl import logger
-from ads.opctl.operator.lowcode.anomaly.const import OutputColumns, SupportedMetrics
+from ads.opctl.operator.lowcode.anomaly.const import OutputColumns, SupportedMetrics, SUBSAMPLE_THRESHOLD
 from ads.opctl.operator.lowcode.anomaly.utils import _build_metrics_df, default_signer
 from ads.opctl.operator.lowcode.common.utils import (
     disable_print,
@@ -79,7 +79,7 @@ def generate_report(self):
                 anomaly_output, test_data, elapsed_time
             )
         table_blocks = [
-            rc.DataTable(df, label=col, index=True)
+            rc.DataTable(df.head(SUBSAMPLE_THRESHOLD) if self.spec.subsample_report_data and len(df) > SUBSAMPLE_THRESHOLD else df, label=col, index=True)
             for col, df in self.datasets.full_data_dict.items()
         ]
         data_table = rc.Select(blocks=table_blocks)
@@ -94,20 +94,36 @@ def generate_report(self):
             anomaly_col = anomaly_output.get_anomalies_by_cat(category=target)[
                 OutputColumns.ANOMALY_COL
             ]
+            anomaly_indices = [i for i, index in enumerate(anomaly_col) if index == 1]
+            downsampled_time_col = time_col
+            selected_indices = list(range(len(time_col)))
+            if self.spec.subsample_report_data:
+                non_anomaly_indices = [i for i in range(len(time_col)) if i not in anomaly_indices]
+                # Downsample non-anomalous data if it exceeds the threshold (1000)
+                if len(non_anomaly_indices) > SUBSAMPLE_THRESHOLD:
+                    downsampled_non_anomaly_indices = non_anomaly_indices[::len(non_anomaly_indices)//SUBSAMPLE_THRESHOLD]
+                    selected_indices = anomaly_indices + downsampled_non_anomaly_indices
+                    selected_indices.sort()
+                downsampled_time_col = time_col[selected_indices]
+
             columns = set(df.columns).difference({date_column})
             for col in columns:
                 y = df[col].reset_index(drop=True)
+
+                downsampled_y = y[selected_indices]
+
                 fig, ax = plt.subplots(figsize=(8, 3), layout="constrained")
                 ax.grid()
-                ax.plot(time_col, y, color="black")
-                for i, index in enumerate(anomaly_col):
-                    if index == 1:
-                        ax.scatter(time_col[i], y[i], color="red", marker="o")
+                ax.plot(downsampled_time_col, downsampled_y, color="black")
+                # Plot anomalies
+                for i in anomaly_indices:
+                    ax.scatter(time_col[i], y[i], color="red", marker="o")
                 plt.xlabel(date_column)
                 plt.ylabel(col)
                 plt.title(f"`{col}` with reference to anomalies")
                 figure_blocks.append(rc.Widget(ax))
-            blocks.append(rc.Group(*figure_blocks, label=target))
+
+        blocks.append(rc.Group(*figure_blocks, label=target))
         plots = rc.Select(blocks)
 
         report_sections = []

ads/opctl/operator/lowcode/anomaly/operator_config.py

@@ -77,6 +77,7 @@ class AnomalyOperatorSpec(DataClassSerializable):
     model: str = None
     model_kwargs: Dict = field(default_factory=dict)
     contamination: float = None
+    subsample_report_data: bool = None
 
     def __post_init__(self):
         """Adjusts the specification details."""

ads/opctl/operator/lowcode/anomaly/schema.yaml

@@ -377,4 +377,8 @@ spec:
       type: dict
       required: false
 
+    subsample_report_data:
+      type: boolean
+      required: false
+
   type: dict

ads/pipeline/ads_pipeline_run.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8; -*-
 
-# Copyright (c) 2022, 2023 Oracle and/or its affiliates.
+# Copyright (c) 2022, 2024 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 import copy
 import logging
@@ -689,6 +689,16 @@ def _build_filter_expression(self, steps: List = [], log_type: str = None) -> st
         sources = []
         subjects = []
         skipped_step_list = []
+
+        is_service_logging_enabled = False
+        try:
+            if self.service_logging:
+                is_service_logging_enabled = True
+        except LogNotConfiguredError:
+            logger.warning(
+                "Service log is not configured for pipeline. Streaming custom log."
+            )
+
         for step_run in self.step_runs:
             if not steps or (step_run.step_name in steps):
                 step_name = step_run.step_name
@@ -703,7 +713,8 @@ def _build_filter_expression(self, steps: List = [], log_type: str = None) -> st
                         subjects.append(f"subject = '{step_name}'")
                     else:
                         sources.append(f"source = '*{job_run_id}'")
-                        subjects.append(f"subject = '{step_name}'")
+                        if is_service_logging_enabled:
+                            subjects.append(f"subject = '{step_name}'")
                 else:
                     subjects.append(f"subject = '{step_name}'")
 

tests/unitary/default_setup/pipeline/test_pipeline_run.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Copyright (c) 2023 Oracle and/or its affiliates.
+# Copyright (c) 2024 Oracle and/or its affiliates.
 # Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
 
 from datetime import datetime
@@ -391,6 +391,8 @@ def test_stream_log(self, mock_stream, mock_sync):
             **PIPELINE_RUN_LOG_DETAILS
         )
         pipeline_run.time_accepted = datetime.now()
+        service_logging = OCILog()
+        pipeline_run._set_service_logging_resource(service_logging)
         pipeline_run._PipelineRun__stream_log(
             ConsolidatedLog(OCILog()),
             [custom_script_step.step_name, ml_job_step.step_name],
@@ -749,6 +751,9 @@ def test_build_filter_expression(self):
             ml_job_step_without_job_run_id,
         ]
 
+        service_logging = OCILog()
+        pipeline_run._set_service_logging_resource(service_logging)
+
         consolidated_log_expression = pipeline_run._build_filter_expression()
 
         assert (