shunping commented on code in PR #34018: URL: https://github.com/apache/beam/pull/34018#discussion_r1962833576
########## sdks/python/apache_beam/ml/anomaly/aggregations.py: ########## @@ -0,0 +1,263 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 collections +import math +import statistics +from typing import Callable +from typing import Iterable + +from apache_beam.ml.anomaly.base import AggregationFn +from apache_beam.ml.anomaly.base import AnomalyPrediction +from apache_beam.ml.anomaly.specifiable import specifiable + + +class LabelAggregation(AggregationFn): + """Aggregates anomaly predictions based on their labels. + + This is an abstract base class for `AggregationFn`s that combine multiple + `AnomalyPrediction` objects into a single `AnomalyPrediction` based on + the labels of the input predictions. + + Args: + agg_func (Callable[[Iterable[int]], int]): A function that aggregates + a collection of anomaly labels (integers) into a single label. + include_history (bool): If True, include the input predictions in the + `agg_history` of the output. Defaults to False. + """ + def __init__( + self, + agg_func: Callable[[Iterable[int]], int], + include_history: bool = False): + self._agg = agg_func + self._include_history = include_history + self._agg_model_id = None + + def apply( + self, predictions: Iterable[AnomalyPrediction]) -> AnomalyPrediction: + """Applies the label aggregation function to a list of predictions. + + Args: + predictions (Iterable[AnomalyPrediction]): A collection of + `AnomalyPrediction` objects to be aggregated. + + Returns: + AnomalyPrediction: A single `AnomalyPrediction` object with the + aggregated label. + """ + labels = [ + prediction.label for prediction in predictions + if prediction.label is not None + ] + + if len(labels) == 0: + return AnomalyPrediction(model_id=self._agg_model_id) + + label = self._agg(labels) + + history = list(predictions) if self._include_history else None + + return AnomalyPrediction( + model_id=self._agg_model_id, label=label, agg_history=history) + + +class ScoreAggregation(AggregationFn): + """Aggregates anomaly predictions based on their scores. + + This is an abstract base class for `AggregationFn`s that combine multiple + `AnomalyPrediction` objects into a single `AnomalyPrediction` based on + the scores of the input predictions. + + Args: + agg_func (Callable[[Iterable[float]], float]): A function that aggregates + a collection of anomaly scores (floats) into a single score. + include_history (bool): If True, include the input predictions in the + `agg_history` of the output. Defaults to False. + """ + def __init__( + self, + agg_func: Callable[[Iterable[float]], float], + include_history: bool = False): + self._agg = agg_func + self._include_history = include_history + self._agg_model_id = None + + def apply( + self, predictions: Iterable[AnomalyPrediction]) -> AnomalyPrediction: + """Applies the score aggregation function to a list of predictions. + + Args: + predictions (Iterable[AnomalyPrediction]): A collection of + `AnomalyPrediction` objects to be aggregated. + + Returns: + AnomalyPrediction: A single `AnomalyPrediction` object with the + aggregated score. + """ + scores = [ + prediction.score for prediction in predictions + if prediction.score is not None and not math.isnan(prediction.score) Review Comment: Thanks, I think you raised a good point here. Take a step back. The basic flow for anomaly detection we implement here is as follows: [input] -> detector -> [score] -> threhold_fn -> [label] -> aggregation_fn -> [aggregated label] - First, I think the situation we are considering here is when a detector generates score of `None` and `NaN`. In my opinion, we can explicitly put up two scenarios when these scores are generated: - The detector is NOT ready to give a prediction. This could imply that the detector needs some warm-up time before the first inference can be made. - The detector is ready to predict, but there is something wrong when it tries to do the prediction. For example, it could be the input data is in ill-format or the detector is simply not able to make a prediction on this input. We can use `None` to represent the first case, and `NaN` for the second one. The rationale is that `None` value is something we don't know yet, but it could be recoverable (if we feed the input into the detector that is ready to score), but `NaN` is coming from an error during prediction and can never be recovered. - After we have `None` and `NaN` scores, the threshold_fn needs to handle how to assign labels for them. - In the current implementation, I only consider `None` and assign a normal label to it, which may be ok, because I don't want to flag outliers when the model is still warming up. Alternatively, we can also set the label to be `None` which means that I will defer the decision to other detectors. - For the irrecoverable `NaN` score, I think we can assign an outlier label. - When multiple labels are ready for aggregation, it is reasonable to apply the aggregation_fn on all the non-None labels. If they are all `None`, maybe we can expose another parameter in the aggregation function for undecided default. WDYT? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
