Exposing Helpful Anomaly Detection Metadata from Anomaly Strategies (ie Anomaly Check Range/Thresholds) through backwards compatible function#593
Conversation
* Add Spark 3.5 support * Replace with DataTypeUtils.fromAttributes * Remove unintended new line
…mpleteness (awslabs#532) * Modified Completeness analyzer to label filtered rows as null for row-level results * Modified GroupingAnalyzers and Uniqueness analyzer to label filtered rows as null for row-level results * Adjustments for modifying the calculate method to take in a filterCondition * Add RowLevelFilterTreatement trait and object to determine how filtered rows will be labeled (default True) * Modify VerificationRunBuilder to have RowLevelFilterTreatment as variable instead of extending, create RowLevelAnalyzer trait * Do row-level filtering in AnalyzerOptions rather than with RowLevelFilterTreatment trait * Modify computeStateFrom to take in optional filterCondition
…elOperations is not available (awslabs#536)
…um (awslabs#535) * Address comments on PR awslabs#532 * Add filtered row-level result support for Minimum, Maximum, Compliance, PatternMatch, MinLength, MaxLength analyzers * Refactored criterion for MinLength and MaxLength analyzers to separate rowLevelResults logic
…wslabs#537) * Add analyzerOption to add filteredRowOutcome for isPrimaryKey Check * Add analyzerOption to add filteredRowOutcome for hasUniqueValueRatio Check
…vior.EmptyString option, the where filter wasn't properly applied (awslabs#538)
…slabs#543) * [Min/Max] Apply filtered row behavior at the row level evaluation - This changes from applying the behavior at the analyzer level. It allows us to prevent the usage of MinValue/MaxValue as placeholder values for filtered rows. * Improved the separation of null rows, based on their source - Whether the outcome for a row is null because of being filtered out or due to the target column being null, is now stored in the outcome column itself. - We could have reused the placeholder value to find out if a row was originally filtered out, but that would not work if the actual value in the row was the same originally. * Mark filtered rows as true We recently fixed the outcome of filtered rows and made them default to true instead of false, which was a bug earlier. This change maintains that behavior. * Added null behavior - empty string to match block Not having it can cause match error.
…aluation (awslabs#547) * [MinLength/MaxLength] Apply filtered row behavior at the row level evaluation - For certain scenarios, the filtered row behavior for MinLength and MaxLength was not working correctly. - For example, when using both minLength and maxLength constraints in a single check, and with both using == <value> as an assertion. This was resulting in the row level outcome of the filtered rows to be false. This was because we were replacing values for filtered rows for Min to MaxValue and for Max to MinValue. But a number could not equal both at the same time. - Updated the logic of the row level assertion to MinLength/MaxLength to match what was done for Min/Max.
- The satisfies constraint was incorrectly using the provided assertion to evaluate the row level outcomes. The assertion should only be used to evaluate the final outcome. - As part of this change, we have updated the row level results to return a true/false. The cast to an integer happens as part of the aggregation result. - Added a test to verify the row level results using checks made up of different assertions.
* Added RatioOfSums analyzer and tests * Unit test for divide by zero and code cleanup. * More detailed Scaladoc * Fixed docs to include Double.NegativeInfinity * Add copyright to new file
* Fix flaky KLL test * Move CustomSql state to CustomSql analyzer * Implement new Analyzer to count columns * Improve documentation, remove unused parameter, replace if/else with map --------- Co-authored-by: Yannis Mentekidis <mentekid@amazon.com>
* Configurable RetainCompletenessRule * Add doc string * Add default completeness const
…awslabs#569) Co-authored-by: Tyler Mcdaniel <tymcd@amazon.com>
* Configurable RetainCompletenessRule * Add doc string * Add default completeness const * Add ConfidenceIntervalStrategy * Add Separate Wilson and Wald Interval Test * Add License information, Fix formatting * Add License information * formatting fix * Update documentation * Make WaldInterval the default strategy for now * Formatting import to per line * Separate group import to per line import
* Add support for EntityTypes dqdl rule * Add support for Conditional Aggregation Analyzer --------- Co-authored-by: Joshua Zexter <jzexter@amazon.com>
* Generate row-level results with withColumns Iteratively using withColumn (singular) causes performance issues when iterating over a large sequence of columns. * Add back UNIQUENESS_ID
… for the bound value and isThresholdInclusive, also adding anomaly detection with extended results README, and adding anomaly detection test with 2 anomaly checks on the same suite
|
@arsenalgunnershubert777 Thanks for the PR! Please allow for additional time to review the PR given the amount of changes. In the meantime, would it be possible to rebase your branch off of latest |
|
Hey @rdsharma26 thanks for reaching out. I think I have some issues with the conflicts (and I also can improve my understanding of Git) |
|
Thanks @arsenalgunnershubert777 |
|
@rdsharma26 got it yea when I run that it has more conflicts, and I notice some classes have reverted to previous states |
|
Hey @rdsharma26 hope you are well, just checking in if you or anyone else has any comments so far. After your review/comments, I will also push another branch to have less commits and create another PR. |
|
@arsenalgunnershubert777 We will review this PR this week. Thank you for the patience. |
rdsharma26
left a comment
There was a problem hiding this comment.
Thanks again for this PR. It is detailed and contains extensive documentation and testing.
I've taken a first pass at the changes and left some comments. I see opportunities to reduce repetitive code by reducing the number of new case classes that are being introduced. The new APIs being introduced look good. Can you have a look at my feedback and send another revision? Thank you!
| def detectWithExtendedResults( | ||
| dataSeries: Vector[Double], | ||
| searchInterval: (Int, Int) = (0, Int.MaxValue)): Seq[(Int, AnomalyDetectionDataPoint)] |
There was a problem hiding this comment.
Can this be part of AnomalyDetectionStrategy ? Do we require a new trait ?
| } | ||
| } | ||
|
|
||
| case class AnomalyDetectorWithExtendedResults(strategy: AnomalyDetectionStrategyWithExtendedResults) { |
There was a problem hiding this comment.
Same question here - Can the methods inside this case class be a part of the AnomalyDetector case class? The methods are new, so it should maintain backwards compatibility. I see repeated code in both sets of methods, so having the 4 methods in one class can help with reducing that repetition. The one challenge will be folding AnomalyDetectionStrategy and AnomalyDetectionStrategyWithExtendedResults into class.
| detectWithExtendedResults(dataSeries, searchInterval) | ||
| .filter { case (_, anomDataPoint) => anomDataPoint.isAnomaly } | ||
| .map { case (i, anomDataPoint) => | ||
| (i, Anomaly(Some(anomDataPoint.dataMetricValue), anomDataPoint.confidence, anomDataPoint.detail)) | ||
| } |
There was a problem hiding this comment.
This logic is repeated across multiple classes. Is it possible to consolidate this logic in one place?
| * along with the AnomalyDetectionExtendedResult object which contains the | ||
| * anomaly detection extended result details. | ||
| */ | ||
| private[deequ] def isNewestPointNonAnomalousWithExtendedResults[S <: State[S]]( |
There was a problem hiding this comment.
Should we change the name of this method? The is<> implies returning a boolean. Since we are returning a case class, perhaps we can update the method name to imply that.
| @@ -1355,4 +1399,117 @@ object Check { | |||
|
|
|||
| detectedAnomalies.anomalies.isEmpty | |||
There was a problem hiding this comment.
In a future PR, perhaps we can remove the logic in this method and rely solely on the new method since it returns the extended result which contains the boolean result.
|
@rdsharma26 thanks so much for taking a look! I will look into these comments soon |
|
@arsenalgunnershubert777 Let us know if you had a chance to look at the comments and if you can submit an update to the PR. If you do so, be sure to sync and rebase against the tracking branch to avoid conflicts. |
| .slice(searchStart, searchEnd) | ||
| .filter { case (value, _) => value < lowerBound || value > upperBound } | ||
| .map { case (value, index) => | ||
|
|
There was a problem hiding this comment.
Bug: detectWithExtendedResults still has .filter { case (value, _) => value < lowerBound || value > upperBound } before the .map, so it only returns anomalous data points, not all data points. This means the method never returns non-anomalous points, breaking the contract of detectWithExtendedResults which should return all data points with an isAnomaly flag. Remove the .filter line.
| @@ -149,7 +170,11 @@ case class OnlineNormalStrategy( | |||
| val detail = Some(s"[OnlineNormalStrategy]: Value ${dataSeries(index)} is not in " + | |||
There was a problem hiding this comment.
Bug: detail is always set (even for non-anomalies) with the "is not in bounds" message. For non-anomalous points, detail should be None. You should conditionally set detail based on calcRes.isAnomaly, similar to how other strategies do it.
| * AnomalyDetectionAssertionResult Class | ||
| * This class is returned by the assertion function Check.isNewestPointNonAnomalousWithExtendedResults. | ||
| * @param hasAnomaly Boolean indicating if there was an anomaly detected. | ||
| * @param anomalyDetectionExtendedResult AnomalyDetectionExtendedResults class. |
There was a problem hiding this comment.
Typo: AnomalyDetectionAssertionResult should be AnomalyDetectionAssertionResult → AnomalyDetectionAssertionResult. Actually the word "Assertion" is misspelled throughout as "Assertion". This affects class names, method names, and variable names across multiple files (AnomalyDetectionAssertionResult, ConstraintAssertionException, AssertionException, anomalyAssertionFunction, etc.). Consider renaming to AnomalyDetectionAssertionResult → AnomalyDetectionAssertionResult... wait, the correct English word is "Assertion", not "Assertion". Rename all occurrences.
|
|
||
| def canEqual(that: Any): Boolean = { | ||
| that.isInstanceOf[AnomalyDetectionDataPoint] | ||
| } |
There was a problem hiding this comment.
AnomalyDetectionDataPoint is a class with a companion object providing apply. Consider making it a case class instead, which would give you equals, hashCode, toString, copy, and pattern matching for free, and remove the manually implemented equals/hashCode/canEqual methods.
| "anomalyDetectionDataPoints from AnomalyDetectionExtendedResult cannot be empty") | ||
|
|
||
| // Get the last anomaly detection data point of sequence (there should only be one element for now). | ||
| // Check the isAnomaly boolean, also return the last anomaly detection data point |
There was a problem hiding this comment.
The match is non-exhaustive — it only handles _ :+ lastAnomalyDataPointPair but not the empty sequence case. Although there's a require above, the compiler doesn't know that, and if the require is somehow bypassed you'll get a MatchError. Add a default case or use .last directly after the require.
| @@ -89,15 +114,25 @@ trait BaseChangeStrategy | |||
| val startPoint = Seq(start - order, 0).max | |||
There was a problem hiding this comment.
The diff function with order can produce fewer elements than dataSeries.slice(startPoint, end). The data.zipWithIndex.map then uses index + startPoint + order to index back into dataSeries. Verify this doesn't go out of bounds when startPoint is 0 and order > start. The original code had the same logic, but now that you return all points (not just anomalies), edge cases may surface more easily.
| * @param hint A hint to provide additional context why a constraint could have failed. | ||
| */ | ||
| def anomalyConstraintWithExtendedResults[S <: State[S]]( | ||
| analyzer: Analyzer[S, Metric[Double]], |
There was a problem hiding this comment.
The Scaladoc says "Runs Completeness analysis" but this is a generic anomaly constraint factory method, not specific to Completeness. Fix the doc comment.
| assert(value < lowerBound || value > upperBound) | ||
| } | ||
| } | ||
|
|
There was a problem hiding this comment.
The "Simple Threshold Strategy with Extended Results" block is nested inside the "Simple Threshold Strategy" should block. This means it's a test case name, not a separate should block. Move it outside to be a sibling should block for proper test organization.
| case nc: ConstraintDecorator => nc.inner | ||
| case c: Constraint => c | ||
| } | ||
| finalConstraint match { |
There was a problem hiding this comment.
This will throw a MatchError if finalConstraint is neither AnalysisBasedConstraint nor AnomalyExtendedResultsConstraint (e.g., a NamedConstraint wrapping something else, or any other Constraint type). Add a default case with a meaningful error.
| val (detail, isAnomaly) = if (anomalyMetricValue > upperBound) { | ||
| (Some(s"Forecasted $forecastedValue for observed value $inputValue"), true) | ||
| } else { | ||
| (None, false) |
There was a problem hiding this comment.
The lowerBound for anomalyCheckRange is set to Double.MinValue which is the most negative finite Double, not negative infinity. Since there's effectively no lower bound check in HoltWinters (only math.abs(inputValue - forecastedValue) > 1.96 * residualSD), using Bound(0.0, inclusive = true) for the lower bound and Bound(upperBound, inclusive = false) for the upper bound would more accurately represent the actual check being performed (absolute difference > threshold).
| .slice(searchStart, searchEnd) | ||
| .filter { case (value, _) => value < lowerBound || value > upperBound } | ||
| .map { case (value, index) => | ||
|
|
There was a problem hiding this comment.
Bug: detectWithExtendedResults still has .filter { case (value, _) => value < lowerBound || value > upperBound } before the .map, so it only returns anomalous data points, not all data points. This breaks the contract of detectWithExtendedResults which should return all data points with an isAnomaly flag (as the other strategies do). Remove the .filter line.
| @@ -149,7 +170,11 @@ case class OnlineNormalStrategy( | |||
| val detail = Some(s"[OnlineNormalStrategy]: Value ${dataSeries(index)} is not in " + | |||
There was a problem hiding this comment.
Bug: detail is always computed with the "is not in bounds" message regardless of whether the point is an anomaly. For non-anomalous points, detail should be None. Wrap the detail in a conditional on calcRes.isAnomaly, similar to how BatchNormalStrategy and BaseChangeStrategy do it.
| "anomalyDetectionDataPoints from AnomalyDetectionExtendedResult cannot be empty") | ||
|
|
||
| // Get the last anomaly detection data point of sequence (there should only be one element for now). | ||
| // Check the isAnomaly boolean, also return the last anomaly detection data point |
There was a problem hiding this comment.
The match is non-exhaustive — it only handles _ :+ lastAnomalyDataPointPair but not the empty sequence case (Nil/Seq()). Although there's a require above, the compiler doesn't know that and you'll get a MatchError at runtime if the require is somehow bypassed. Add a default case or use .last directly after the require:
val lastPair = extendedDetectionResult.anomalyDetectionDataPointSequence.last
(lastPair._2.isAnomaly, AnomalyDetectionExtendedResult(lastPair._2))| new NamedConstraint(constraint, s"AnomalyConstraint($analyzer)") | ||
| } | ||
|
|
||
| /** |
There was a problem hiding this comment.
Scaladoc says "Runs Completeness analysis on the given column" but this is a generic anomaly constraint factory method, not specific to Completeness. Fix the doc comment.
|
This PR has been inactive for 60 days. It will be closed in 14 days if there is no further activity. If you are still working on this, please push an update or comment to keep it open. |
Description of changes:
This PR adds functionality to expose anomaly detection metadata for anomaly checks.
There is a legacy PR here, and this PR addresses feedback to make the anomaly detection metadata functionality backwards compatible.
Let me know if anyone has any questions/feedback, thanks!
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.