Skip to content

Exposing Helpful Anomaly Detection Metadata from Anomaly Strategies (ie Anomaly Check Range/Thresholds) through backwards compatible function#593

Open
arsenalgunnershubert777 wants to merge 25 commits into
awslabs:masterfrom
arsenalgunnershubert777:exposeAnomalyThresholdBackwardsCompatible
Open

Exposing Helpful Anomaly Detection Metadata from Anomaly Strategies (ie Anomaly Check Range/Thresholds) through backwards compatible function#593
arsenalgunnershubert777 wants to merge 25 commits into
awslabs:masterfrom
arsenalgunnershubert777:exposeAnomalyThresholdBackwardsCompatible

Conversation

@arsenalgunnershubert777

Copy link
Copy Markdown

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.

* 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
…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
* 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
@rdsharma26

Copy link
Copy Markdown
Contributor

@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 master so that it reduces the number of commits in the PR?

@arsenalgunnershubert777

Copy link
Copy Markdown
Author

Hey @rdsharma26 thanks for reaching out. I think I have some issues with the conflicts (and I also can improve my understanding of Git)
So for now I may just create a clean branch and apply the changes on top of that manually and open a new PR, so there's just 1 commit

@rdsharma26

rdsharma26 commented Nov 13, 2024

Copy link
Copy Markdown
Contributor

Thanks @arsenalgunnershubert777
After syncing your forked repository with the main repository, you could try:

git checkout master
git pull
git checkout exposeAnomalyThresholdBackwardsCompatible
git rebase master
git push origin exposeAnomalyThresholdBackwardsCompatible

@arsenalgunnershubert777

Copy link
Copy Markdown
Author

@rdsharma26 got it yea when I run that it has more conflicts, and I notice some classes have reverted to previous states

@arsenalgunnershubert777

arsenalgunnershubert777 commented Dec 11, 2024

Copy link
Copy Markdown
Author

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.

@rdsharma26

Copy link
Copy Markdown
Contributor

@arsenalgunnershubert777 We will review this PR this week. Thank you for the patience.

@rdsharma26 rdsharma26 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@arsenalgunnershubert777

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!

Comment on lines +43 to +45
def detectWithExtendedResults(
dataSeries: Vector[Double],
searchInterval: (Int, Int) = (0, Int.MaxValue)): Seq[(Int, AnomalyDetectionDataPoint)]

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can this be part of AnomalyDetectionStrategy ? Do we require a new trait ?

}
}

case class AnomalyDetectorWithExtendedResults(strategy: AnomalyDetectionStrategyWithExtendedResults) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +235 to +239
detectWithExtendedResults(dataSeries, searchInterval)
.filter { case (_, anomDataPoint) => anomDataPoint.isAnomaly }
.map { case (i, anomDataPoint) =>
(i, Anomaly(Some(anomDataPoint.dataMetricValue), anomDataPoint.confidence, anomDataPoint.detail))
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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]](

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@arsenalgunnershubert777

Copy link
Copy Markdown
Author

@rdsharma26 thanks so much for taking a look! I will look into these comments soon

@rdsharma26

Copy link
Copy Markdown
Contributor

@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.

@github-actions github-actions Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Generated by AI (model: us.anthropic.claude-opus-4-6-v1, prompt: 926c07a3) — may not be fully accurate. Reply if this doesn't help.

.slice(searchStart, searchEnd)
.filter { case (value, _) => value < lowerBound || value > upperBound }
.map { case (value, index) =>

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 " +

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Typo: AnomalyDetectionAssertionResult should be AnomalyDetectionAssertionResultAnomalyDetectionAssertionResult. 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 AnomalyDetectionAssertionResultAnomalyDetectionAssertionResult... wait, the correct English word is "Assertion", not "Assertion". Rename all occurrences.


def canEqual(that: Any): Boolean = {
that.isInstanceOf[AnomalyDetectionDataPoint]
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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]],

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)
}
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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).

@github-actions github-actions Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.


Generated by AI (model: us.anthropic.claude-opus-4-6-v1, prompt: 38d3f74c) — may not be fully accurate. Reply if this doesn't help.

.slice(searchStart, searchEnd)
.filter { case (value, _) => value < lowerBound || value > upperBound }
.map { case (value, index) =>

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 " +

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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)")
}

/**

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@github-actions

github-actions Bot commented Jul 6, 2026

Copy link
Copy Markdown
Contributor

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.

@github-actions github-actions Bot added the stale label Jul 6, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.