FIX: mad ad now reliably returns an empty spark frame if no anomalies are detected

Signed-off-by: Christoph Huy <christoph.huy@campus.tu-berlin.de>

Author: Perimora

src/sdk/python/rtdip_sdk/pipelines/anomaly_detection/spark/mad/mad_anomaly_detection.py

@@ -15,6 +15,7 @@
 import pandas as pd
 
 from pyspark.sql import DataFrame
+from pyspark.sql.types import StructField, StructType, DoubleType, BooleanType
 from typing import Optional, List, Union
 
 from ...._pipeline_utils.models import (
@@ -210,6 +211,22 @@ def libraries() -> Libraries:
     def settings() -> dict:
         return {}
 
+    @staticmethod
+    def _build_result_schema(df: DataFrame) -> StructType:
+        return StructType(
+            list(df.schema.fields)
+            + [
+                StructField("mad_zscore", DoubleType(), True),
+                StructField("is_anomaly", BooleanType(), True),
+            ]
+        )
+
+    @staticmethod
+    def _empty_result_df(df: DataFrame, schema: StructType) -> DataFrame:
+        """Create an empty DataFrame with the correct schema using pandas."""
+        empty_pdf = pd.DataFrame(columns=schema.fieldNames())
+        return df.sparkSession.createDataFrame(empty_pdf, schema=schema)
+
     def detect(self, df: DataFrame) -> DataFrame:
         """
         Detects anomalies in the input DataFrame using the configured MAD scorer.
@@ -228,13 +245,25 @@ def detect(self, df: DataFrame) -> DataFrame:
                 - `is_anomaly`: Boolean anomaly flag.
         """
 
+        result_schema = self._build_result_schema(df)
+
         pdf = df.toPandas()
+        if pdf.empty:
+            return self._empty_result_df(df, result_schema)
 
         scores = self.scorer.score(pdf["value"])
         pdf["mad_zscore"] = scores
         pdf["is_anomaly"] = self.scorer.is_anomaly(scores)
 
-        return df.sparkSession.createDataFrame(pdf[pdf["is_anomaly"]].copy())
+        anomalies_pdf = pdf[pdf["is_anomaly"]].copy()
+        anomalies_pdf = anomalies_pdf[result_schema.fieldNames()]
+
+        if anomalies_pdf.empty:
+            return self._empty_result_df(df, result_schema)
+
+        # Ensure correct column order matches schema
+        anomalies_pdf = anomalies_pdf[result_schema.fieldNames()]
+        return df.sparkSession.createDataFrame(anomalies_pdf, schema=result_schema)
 
 
 class DecompositionMadAnomalyDetection(AnomalyDetectionInterface):
@@ -368,6 +397,16 @@ def _decompose(self, df: DataFrame) -> DataFrame:
         else:
             raise ValueError(f"Unsupported decomposition method: {self.decomposition}")
 
+    @staticmethod
+    def _build_result_schema(df: DataFrame) -> StructType:
+        return StructType(
+            list(df.schema.fields)
+            + [
+                StructField("mad_zscore", DoubleType(), True),
+                StructField("is_anomaly", BooleanType(), True),
+            ]
+        )
+
     def detect(self, df: DataFrame) -> DataFrame:
         """
         Detects anomalies by scoring the decomposition residuals using the configured MAD scorer.
@@ -385,12 +424,25 @@ def detect(self, df: DataFrame) -> DataFrame:
                 - `mad_zscore`: MAD-based anomaly score computed on `residual`.
                 - `is_anomaly`: Boolean anomaly flag.
         """
-        
+
         decomposed_df = self._decompose(df)
+        result_schema = self._build_result_schema(decomposed_df)
+
         pdf = decomposed_df.toPandas().sort_values(self.timestamp_column)
 
+        if pdf.empty:
+            return MadAnomalyDetection._empty_result_df(decomposed_df, result_schema)
+
         scores = self.scorer.score(pdf["residual"])
         pdf["mad_zscore"] = scores
         pdf["is_anomaly"] = self.scorer.is_anomaly(scores)
 
-        return df.sparkSession.createDataFrame(pdf[pdf["is_anomaly"]].copy())
+        anomalies_pdf = pdf[pdf["is_anomaly"]].copy()
+        anomalies_pdf = anomalies_pdf[result_schema.fieldNames()]
+
+        if anomalies_pdf.empty:
+            return MadAnomalyDetection._empty_result_df(decomposed_df, result_schema)
+
+        # Ensure correct column order matches schema
+        anomalies_pdf = anomalies_pdf[result_schema.fieldNames()]
+        return df.sparkSession.createDataFrame(anomalies_pdf, schema=result_schema)

tests/sdk/python/rtdip_sdk/pipelines/anomaly_detection/spark/test_mad.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import numpy as np
+import pandas as pd
 import pytest
 
 from src.sdk.python.rtdip_sdk.pipelines.anomaly_detection.spark.mad.mad_anomaly_detection import (
@@ -22,6 +24,21 @@
 )
 
 
+@pytest.fixture
+def spark_dataframe_without_anomalies(spark_session):
+    data = [(i, float(10.0 + 0.05 * np.sin(i))) for i in range(1, 31)]
+    columns = ["timestamp", "value"]
+    return spark_session.createDataFrame(data, columns)
+
+
+@pytest.fixture
+def spark_dataframe_without_anomalies_timestamp(spark_session):
+    timestamps = pd.date_range("2025-02-01", periods=72, freq="h")
+    values = 10.0 + 0.1 * np.sin(np.arange(72))
+    pdf = pd.DataFrame({"timestamp": timestamps, "value": values})
+    return spark_session.createDataFrame(pdf)
+
+
 @pytest.fixture
 def spark_dataframe_with_anomalies(spark_session):
     data = [
@@ -40,6 +57,28 @@ def spark_dataframe_with_anomalies(spark_session):
     return spark_session.createDataFrame(data, columns)
 
 
+def test_mad_anomaly_detection_global_no_anomalies(
+    spark_dataframe_without_anomalies,
+):
+    mad_detector = MadAnomalyDetection()
+
+    result_df = mad_detector.detect(spark_dataframe_without_anomalies)
+
+    assert result_df.count() == 0
+    assert result_df.columns == ["timestamp", "value", "mad_zscore", "is_anomaly"]
+
+
+def test_mad_anomaly_detection_rolling_no_anomalies(
+    spark_dataframe_without_anomalies,
+):
+    mad_detector = MadAnomalyDetection(scorer=RollingMadScorer(window_size=5))
+
+    result_df = mad_detector.detect(spark_dataframe_without_anomalies)
+
+    assert result_df.count() == 0
+    assert result_df.columns == ["timestamp", "value", "mad_zscore", "is_anomaly"]
+
+
 def test_mad_anomaly_detection_global(spark_dataframe_with_anomalies):
     mad_detector = MadAnomalyDetection()
 
@@ -136,7 +175,7 @@ def spark_dataframe_synthetic_stl(spark_session):
     n = 500
     period = 24
 
-    timestamps = pd.date_range("2025-01-01", periods=n, freq="H")
+    timestamps = pd.date_range("2025-01-01", periods=n, freq="h")
     trend = 0.02 * np.arange(n)
     seasonal = 5 * np.sin(2 * np.pi * np.arange(n) / period)
     noise = 0.3 * np.random.randn(n)
@@ -151,6 +190,33 @@ def spark_dataframe_synthetic_stl(spark_session):
     return spark_session.createDataFrame(pdf)
 
 
+@pytest.mark.parametrize(
+    "decomposition, scorer",
+    [
+        ("stl", GlobalMadScorer(threshold=3.5)),
+        ("mstl", RollingMadScorer(threshold=3.5, window_size=24)),
+    ],
+)
+def test_decomposition_mad_anomaly_detection_no_anomalies(
+    spark_dataframe_without_anomalies_timestamp,
+    decomposition,
+    scorer,
+):
+    detector = DecompositionMadAnomalyDetection(
+        scorer=scorer,
+        decomposition=decomposition,
+        period=24,
+        timestamp_column="timestamp",
+        value_column="value",
+    )
+
+    result_df = detector.detect(spark_dataframe_without_anomalies_timestamp)
+
+    assert result_df.count() == 0
+    assert "mad_zscore" in result_df.columns
+    assert "is_anomaly" in result_df.columns
+
+
 @pytest.mark.parametrize(
     "decomposition, period, scorer",
     [