You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jr...@apache.org on 2025/04/29 18:25:12 UTC
(beam) branch master updated: Modernize Anomaly Detection type hints (#34779)
This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 681cc3c0ad5 Modernize Anomaly Detection type hints (#34779)
681cc3c0ad5 is described below
commit 681cc3c0ad5d3ca5d5ac4c8f96620bdedc958b75
Author: Jack McCluskey <34...@users.noreply.github.com>
AuthorDate: Tue Apr 29 14:25:06 2025 -0400
Modernize Anomaly Detection type hints (#34779)
---
sdks/python/apache_beam/ml/anomaly/aggregations.py | 4 +--
sdks/python/apache_beam/ml/anomaly/base.py | 7 ++---
.../apache_beam/ml/anomaly/detectors/offline.py | 20 ++++++-------
.../ml/anomaly/detectors/pyod_adapter.py | 7 ++---
sdks/python/apache_beam/ml/anomaly/specifiable.py | 15 ++++------
.../apache_beam/ml/anomaly/specifiable_test.py | 3 +-
sdks/python/apache_beam/ml/anomaly/transforms.py | 34 ++++++++++------------
.../apache_beam/ml/anomaly/transforms_test.py | 5 ++--
8 files changed, 42 insertions(+), 53 deletions(-)
diff --git a/sdks/python/apache_beam/ml/anomaly/aggregations.py b/sdks/python/apache_beam/ml/anomaly/aggregations.py
index 6d9f3797663..35be249879a 100644
--- a/sdks/python/apache_beam/ml/anomaly/aggregations.py
+++ b/sdks/python/apache_beam/ml/anomaly/aggregations.py
@@ -18,9 +18,9 @@
import collections
import math
import statistics
+from collections.abc import Callable
+from collections.abc import Iterable
from typing import Any
-from typing import Callable
-from typing import Iterable
from typing import Optional
from apache_beam.ml.anomaly.base import DEFAULT_MISSING_LABEL
diff --git a/sdks/python/apache_beam/ml/anomaly/base.py b/sdks/python/apache_beam/ml/anomaly/base.py
index 8251245f1ca..991c73ad118 100644
--- a/sdks/python/apache_beam/ml/anomaly/base.py
+++ b/sdks/python/apache_beam/ml/anomaly/base.py
@@ -21,9 +21,8 @@ Base classes for anomaly detection
from __future__ import annotations
import abc
+from collections.abc import Iterable
from dataclasses import dataclass
-from typing import Iterable
-from typing import List
from typing import Optional
import apache_beam as beam
@@ -186,7 +185,7 @@ class EnsembleAnomalyDetector(AnomalyDetector):
"""An abstract base class for an ensemble of anomaly (sub-)detectors.
Args:
- sub_detectors: A List of `AnomalyDetector` used in this ensemble model.
+ sub_detectors: A list of `AnomalyDetector` used in this ensemble model.
aggregation_strategy: An optional `AggregationFn` to apply to the
predictions from all sub-detectors and yield an aggregated result.
model_id: Inherited from `AnomalyDetector`.
@@ -196,7 +195,7 @@ class EnsembleAnomalyDetector(AnomalyDetector):
"""
def __init__(
self,
- sub_detectors: Optional[List[AnomalyDetector]] = None,
+ sub_detectors: Optional[list[AnomalyDetector]] = None,
aggregation_strategy: Optional[AggregationFn] = None,
**kwargs):
if "model_id" not in kwargs or kwargs["model_id"] is None:
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/offline.py b/sdks/python/apache_beam/ml/anomaly/detectors/offline.py
index 2bd939acb68..e66f21eb3a1 100644
--- a/sdks/python/apache_beam/ml/anomaly/detectors/offline.py
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/offline.py
@@ -16,11 +16,9 @@
#
from typing import Any
-from typing import Dict
from typing import Optional
from typing import SupportsFloat
from typing import SupportsInt
-from typing import Tuple
from typing import TypeVar
import apache_beam as beam
@@ -47,8 +45,8 @@ class OfflineDetector(AnomalyDetector):
"""
@staticmethod
def score_prediction_adapter(
- keyed_prediction: Tuple[KeyT, PredictionResult]
- ) -> Tuple[KeyT, AnomalyPrediction]:
+ keyed_prediction: tuple[KeyT, PredictionResult]
+ ) -> tuple[KeyT, AnomalyPrediction]:
"""Extracts a float score from `PredictionResult.inference` and wraps it.
Takes a keyed `PredictionResult` from common ModelHandler output, assumes
@@ -56,11 +54,11 @@ class OfflineDetector(AnomalyDetector):
paired with an `AnomalyPrediction` containing that float score.
Args:
- keyed_prediction: Tuple of `(key, PredictionResult)`. `PredictionResult`
+ keyed_prediction: tuple of `(key, PredictionResult)`. `PredictionResult`
must have an `inference` attribute supporting float conversion.
Returns:
- Tuple of `(key, AnomalyPrediction)` with the extracted score.
+ tuple of `(key, AnomalyPrediction)` with the extracted score.
Raises:
AssertionError: If `PredictionResult.inference` doesn't support float().
@@ -73,8 +71,8 @@ class OfflineDetector(AnomalyDetector):
@staticmethod
def label_prediction_adapter(
- keyed_prediction: Tuple[KeyT, PredictionResult]
- ) -> Tuple[KeyT, AnomalyPrediction]:
+ keyed_prediction: tuple[KeyT, PredictionResult]
+ ) -> tuple[KeyT, AnomalyPrediction]:
"""Extracts an integer label from `PredictionResult.inference` and wraps it.
Takes a keyed `PredictionResult`, assumes its `inference` attribute is an
@@ -82,11 +80,11 @@ class OfflineDetector(AnomalyDetector):
`AnomalyPrediction` containing that integer label.
Args:
- keyed_prediction: Tuple of `(key, PredictionResult)`. `PredictionResult`
+ keyed_prediction: tuple of `(key, PredictionResult)`. `PredictionResult`
must have an `inference` attribute supporting int conversion.
Returns:
- Tuple of `(key, AnomalyPrediction)` with the extracted label.
+ tuple of `(key, AnomalyPrediction)` with the extracted label.
Raises:
AssertionError: If `PredictionResult.inference` doesn't support int().
@@ -100,7 +98,7 @@ class OfflineDetector(AnomalyDetector):
def __init__(
self,
keyed_model_handler: KeyedModelHandler[Any, beam.Row, PredictionT, Any],
- run_inference_args: Optional[Dict[str, Any]] = None,
+ run_inference_args: Optional[dict[str, Any]] = None,
**kwargs):
super().__init__(**kwargs)
diff --git a/sdks/python/apache_beam/ml/anomaly/detectors/pyod_adapter.py b/sdks/python/apache_beam/ml/anomaly/detectors/pyod_adapter.py
index f8df78261b9..10bd2551476 100644
--- a/sdks/python/apache_beam/ml/anomaly/detectors/pyod_adapter.py
+++ b/sdks/python/apache_beam/ml/anomaly/detectors/pyod_adapter.py
@@ -16,11 +16,10 @@
#
import pickle
+from collections.abc import Iterable
+from collections.abc import Sequence
from typing import Any
-from typing import Dict
-from typing import Iterable
from typing import Optional
-from typing import Sequence
import numpy as np
@@ -74,7 +73,7 @@ class PyODModelHandler(ModelHandler[beam.Row,
self,
batch: Sequence[beam.Row],
model: PyODBaseDetector,
- inference_args: Optional[Dict[str, Any]] = None
+ inference_args: Optional[dict[str, Any]] = None
) -> Iterable[PredictionResult]:
np_batch = []
for row in batch:
diff --git a/sdks/python/apache_beam/ml/anomaly/specifiable.py b/sdks/python/apache_beam/ml/anomaly/specifiable.py
index f13c9496672..9cefcbfeabe 100644
--- a/sdks/python/apache_beam/ml/anomaly/specifiable.py
+++ b/sdks/python/apache_beam/ml/anomaly/specifiable.py
@@ -26,13 +26,10 @@ import dataclasses
import inspect
import logging
import os
+from collections.abc import Callable
from typing import Any
-from typing import Callable
-from typing import Dict
-from typing import List
from typing import Optional
from typing import Protocol
-from typing import Type
from typing import TypeVar
from typing import Union
from typing import overload
@@ -64,7 +61,7 @@ T = TypeVar('T', bound=type)
BUILTIN_TYPES_IN_SPEC = (int, float, complex, str, bytes, bytearray)
-def _class_to_subspace(cls: Type) -> str:
+def _class_to_subspace(cls: type) -> str:
"""
Search the class hierarchy to find the subspace: the closest ancestor class in
the class's method resolution order (MRO) whose name is found in the accepted
@@ -103,14 +100,14 @@ class Spec():
#: An optional dictionary of keyword arguments for the `__init__` method of
#: the class. If None, when we materialize this Spec, we only return the
#: class without instantiate any objects from it.
- config: Optional[Dict[str, Any]] = dataclasses.field(default_factory=dict)
+ config: Optional[dict[str, Any]] = dataclasses.field(default_factory=dict)
def _specifiable_from_spec_helper(v, _run_init):
if isinstance(v, Spec):
return Specifiable.from_spec(v, _run_init)
- if isinstance(v, List):
+ if isinstance(v, list):
return [_specifiable_from_spec_helper(e, _run_init) for e in v]
# TODO: support spec treatment for more types
@@ -126,7 +123,7 @@ def _specifiable_to_spec_helper(v):
if isinstance(v, Specifiable):
return v.to_spec()
- if isinstance(v, List):
+ if isinstance(v, list):
return [_specifiable_to_spec_helper(e) for e in v]
if inspect.isfunction(v):
@@ -172,7 +169,7 @@ class Specifiable(Protocol):
raise ValueError(f"Spec type not found in {spec}")
subspace = _spec_type_to_subspace(spec.type)
- subclass: Type[Self] = _KNOWN_SPECIFIABLE[subspace].get(spec.type, None)
+ subclass: type[Self] = _KNOWN_SPECIFIABLE[subspace].get(spec.type, None)
if subclass is None:
raise ValueError(f"Unknown spec type '{spec.type}' in {spec}")
diff --git a/sdks/python/apache_beam/ml/anomaly/specifiable_test.py b/sdks/python/apache_beam/ml/anomaly/specifiable_test.py
index 8ededd2deff..4312b5b1fbf 100644
--- a/sdks/python/apache_beam/ml/anomaly/specifiable_test.py
+++ b/sdks/python/apache_beam/ml/anomaly/specifiable_test.py
@@ -20,7 +20,6 @@ import dataclasses
import logging
import os
import unittest
-from typing import List
from typing import Optional
from parameterized import parameterized
@@ -167,7 +166,7 @@ class TestSpecifiable(unittest.TestCase):
@dataclasses.dataclass
class ShoppingCart():
user_id: str
- entries: List[Entry]
+ entries: list[Entry]
orange = Product("orange", 1.0)
diff --git a/sdks/python/apache_beam/ml/anomaly/transforms.py b/sdks/python/apache_beam/ml/anomaly/transforms.py
index c497421be02..d704f93ed61 100644
--- a/sdks/python/apache_beam/ml/anomaly/transforms.py
+++ b/sdks/python/apache_beam/ml/anomaly/transforms.py
@@ -17,12 +17,10 @@
import dataclasses
import uuid
+from collections.abc import Callable
+from collections.abc import Iterable
from typing import Any
-from typing import Callable
-from typing import Dict
-from typing import Iterable
from typing import Optional
-from typing import Tuple
from typing import TypeVar
import apache_beam as beam
@@ -42,10 +40,10 @@ from apache_beam.transforms.userstate import ReadModifyWriteStateSpec
KeyT = TypeVar('KeyT')
TempKeyT = TypeVar('TempKeyT', bound=int)
-InputT = Tuple[KeyT, beam.Row]
-KeyedInputT = Tuple[KeyT, Tuple[TempKeyT, beam.Row]]
-KeyedOutputT = Tuple[KeyT, Tuple[TempKeyT, AnomalyResult]]
-OutputT = Tuple[KeyT, AnomalyResult]
+InputT = tuple[KeyT, beam.Row]
+KeyedInputT = tuple[KeyT, tuple[TempKeyT, beam.Row]]
+KeyedOutputT = tuple[KeyT, tuple[TempKeyT, AnomalyResult]]
+OutputT = tuple[KeyT, AnomalyResult]
class _ScoreAndLearnDoFn(beam.DoFn):
@@ -60,7 +58,7 @@ class _ScoreAndLearnDoFn(beam.DoFn):
def __init__(self, detector_spec: Spec):
self._detector_spec = detector_spec
- assert isinstance(self._detector_spec.config, Dict)
+ assert isinstance(self._detector_spec.config, dict)
self._detector_spec.config["_run_init"] = True
def score_and_learn(self, data):
@@ -180,7 +178,7 @@ class _StatelessThresholdDoFn(_BaseThresholdDoFn):
creation of a stateful `ThresholdFn`.
"""
def __init__(self, threshold_fn_spec: Spec):
- assert isinstance(threshold_fn_spec.config, Dict)
+ assert isinstance(threshold_fn_spec.config, dict)
threshold_fn_spec.config["_run_init"] = True
self._threshold_fn = Specifiable.from_spec(threshold_fn_spec)
assert isinstance(self._threshold_fn, ThresholdFn)
@@ -191,7 +189,7 @@ class _StatelessThresholdDoFn(_BaseThresholdDoFn):
"""Processes a batch of anomaly results using a stateless ThresholdFn.
Args:
- element (Tuple[Any, Tuple[Any, AnomalyResult]]): A tuple representing
+ element (tuple[Any, tuple[Any, AnomalyResult]]): A tuple representing
an element in the Beam pipeline. It is expected to be in the format
`(key1, (key2, AnomalyResult))`, where key1 is the original input key,
and key2 is a disambiguating key for distinct data points.
@@ -199,7 +197,7 @@ class _StatelessThresholdDoFn(_BaseThresholdDoFn):
in Beam DoFns.
Yields:
- Iterable[Tuple[Any, Tuple[Any, AnomalyResult]]]: An iterable containing
+ Iterable[tuple[Any, tuple[Any, AnomalyResult]]]: An iterable containing
a single output element with the same structure as the input, but with
the `AnomalyResult` having updated prediction labels based on the
stateless `ThresholdFn`.
@@ -227,7 +225,7 @@ class _StatefulThresholdDoFn(_BaseThresholdDoFn):
THRESHOLD_STATE_INDEX = ReadModifyWriteStateSpec('saved_tracker', DillCoder())
def __init__(self, threshold_fn_spec: Spec):
- assert isinstance(threshold_fn_spec.config, Dict)
+ assert isinstance(threshold_fn_spec.config, dict)
threshold_fn_spec.config["_run_init"] = True
threshold_fn = Specifiable.from_spec(threshold_fn_spec)
assert isinstance(threshold_fn, ThresholdFn)
@@ -248,7 +246,7 @@ class _StatefulThresholdDoFn(_BaseThresholdDoFn):
Beam for future elements.
Args:
- element (Tuple[Any, Tuple[Any, AnomalyResult]]): A tuple representing
+ element (tuple[Any, tuple[Any, AnomalyResult]]): A tuple representing
an element in the Beam pipeline. It is expected to be in the format
`(key1, (key2, AnomalyResult))`, where key1 is the original input key,
and key2 is a disambiguating key for distinct data points.
@@ -259,7 +257,7 @@ class _StatefulThresholdDoFn(_BaseThresholdDoFn):
in Beam DoFns.
Yields:
- Iterable[Tuple[Any, Tuple[Any, AnomalyResult]]]: An iterable containing
+ Iterable[tuple[Any, tuple[Any, AnomalyResult]]]: An iterable containing
a single output element with the same structure as the input, but
with the `AnomalyResult` having updated prediction labels based on
the stateful `ThresholdFn`.
@@ -427,7 +425,7 @@ class RunOfflineDetector(beam.PTransform[beam.PCollection[KeyedInputT],
self._offline_detector = offline_detector
def _restore_and_convert(
- self, elem: Tuple[Tuple[Any, Any, beam.Row], Any]) -> KeyedOutputT:
+ self, elem: tuple[tuple[Any, Any, beam.Row], Any]) -> KeyedOutputT:
"""Converts the model output to AnomalyResult.
Args:
@@ -454,8 +452,8 @@ class RunOfflineDetector(beam.PTransform[beam.PCollection[KeyedInputT],
])
return orig_key, (temp_key, result)
- def _select_features(self, elem: Tuple[Any,
- beam.Row]) -> Tuple[Any, beam.Row]:
+ def _select_features(self, elem: tuple[Any,
+ beam.Row]) -> tuple[Any, beam.Row]:
assert self._offline_detector._features is not None
k, v = elem
row_dict = v._asdict()
diff --git a/sdks/python/apache_beam/ml/anomaly/transforms_test.py b/sdks/python/apache_beam/ml/anomaly/transforms_test.py
index 23204d65dbb..f743b1629ee 100644
--- a/sdks/python/apache_beam/ml/anomaly/transforms_test.py
+++ b/sdks/python/apache_beam/ml/anomaly/transforms_test.py
@@ -22,9 +22,8 @@ import pickle
import shutil
import tempfile
import unittest
+from collections.abc import Iterable
from typing import Any
-from typing import Iterable
-from typing import Tuple
import mock
import numpy
@@ -281,7 +280,7 @@ class FakeNumpyModel():
return [input_vector[0][0] * 10 - input_vector[0][1]]
-def _to_keyed_numpy_array(t: Tuple[Any, beam.Row]):
+def _to_keyed_numpy_array(t: tuple[Any, beam.Row]):
"""Converts an Apache Beam Row to a NumPy array."""
return t[0], numpy.array(list(t[1]))