Add DateTimeMetric, Analyzer and Example - #568
Conversation
|
Hi @rdsharma26, @mentekid please help review this one |
eycho-am
left a comment
eycho-am
left a comment
There was a problem hiding this comment.
In general the changes look good and thank you for adding the example.
Could you add some more testing please?
Would like to see more tests for the analyzer behavior (in https://github.com/awslabs/deequ/tree/master/src/test/scala/com/amazon/deequ/analyzers) and tests to see how it works within the VerificationSuite (in https://github.com/awslabs/deequ/blob/master/src/test/scala/com/amazon/deequ/VerificationSuiteTest.scala)
| ) | ||
| } | ||
|
|
||
| "datetimeDistribution analyzer with VerificationSuite" in withSparkSessionJava8APIEnabled { sparkSession => |
There was a problem hiding this comment.
@eycho-am Hi just to confirm this is the kind of test you was looking for right?
Currently It is only implemented as an Analyzer to run with AnalysisRunner or as RequiredAnalyzer
For more advance use case with VerificationSuite, I can open a PR to implement some check based on it
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
|
|
||
| private[this] val dateTypes = Set(TimestampType, DateType) | ||
|
|
||
| private[this] val caseSensitive = { |
There was a problem hiding this comment.
Bug: Set(StructType, MapType, ArrayType) compares companion objects, not types. But dateTypes uses the same pattern with Set(TimestampType, DateType). This set is never actually used for matching (the isDateType method uses pattern matching instead), so it's dead code that only appears in the error message. However, the error message dateTypes.mkString(", ") will print the companion object toString representations, which may not be what you want. Consider using Set("TimestampType", "DateType") or just inline the string.
| @@ -405,6 +435,20 @@ object Preconditions { | |||
| } | |||
| } | |||
There was a problem hiding this comment.
Misleading scaladoc: says /** Specified column has string type */ but the method checks for date/timestamp type.
| case (x, y) => (Instant.ofEpochMilli(x), Instant.ofEpochMilli(x + frequency - 1L)) -> y | ||
| }) | ||
| } | ||
|
|
There was a problem hiding this comment.
The computeStateFromResult computes the end of the interval as x + frequency - 1L (in milliseconds). This means the last millisecond of each bucket overlaps conceptually with the next bucket's start. For example, with DAILY interval, the end is ...T23:59:59.999Z but the next bucket starts at ...T00:00:00.000Z of the next day. This works but is fragile — consider documenting that the range is inclusive on both ends, or use x + frequency as an exclusive upper bound.
| instance: String, | ||
| value: Try[Instant] | ||
| ) extends Metric[Instant] { | ||
| override def flatten(): Seq[DoubleMetric] = value match { |
There was a problem hiding this comment.
DateTimeMetric.flatten() converts to epoch millis as a DoubleMetric. Double has only 53 bits of mantissa, and epoch millis can exceed 2^53 for dates far in the future, causing precision loss. For current dates this is fine, but it's worth documenting this limitation or considering epoch seconds instead.
| * @return | ||
| */ | ||
| def hasPastDates( | ||
| column: String, |
There was a problem hiding this comment.
hasPastDates uses now() which is evaluated at Spark query planning time. If the check is reused or the DataFrame is lazily evaluated, the now() value may differ from expectations. Also, now() is not a standard Spark SQL function — you likely mean current_timestamp().
| * @return | ||
| */ | ||
| def hasFutureDates( | ||
| column: String, |
There was a problem hiding this comment.
Same issue as hasPastDates: now() is not a standard Spark SQL function. Use current_timestamp() instead.
| */ | ||
|
|
||
| package org.apache.spark.sql | ||
|
|
There was a problem hiding this comment.
Package is org.apache.spark.sql but the file lives under com/amazon/deequ/analyzers/catalyst/. While this is intentional to access Spark internals (like other files in this package), the class DateTimeAggregation doesn't seem to need internal Spark access — it only uses public Aggregator API. Consider moving it to the deequ package.
| .master("local") | ||
| .appName("test") | ||
| .config("spark.ui.enabled", "false") | ||
| .config("spark.sql.datetime.java8API.enabled", "true") |
There was a problem hiding this comment.
Setting spark.sql.datetime.java8API.enabled=true globally in ExampleUtils affects all examples, not just the DateTime ones. This could change behavior of existing examples that rely on java.sql.Timestamp/java.sql.Date types.
|
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. |
add null-input NPE guard removed bogus MONTHLY interval changenow() to current_timestamp() Styling changes
There was a problem hiding this comment.
Generated by AI (model: us.anthropic.claude-opus-4-8, prompt: 4d2f5d73) — may not be fully accurate. Reply if this doesn't help.
Reviewed by Shadow · github.com/sudsali/shadow
Additional feedback:
src/main/scala/com/amazon/deequ/analyzers/StateProvider.scala:214 — BUG: HdfsStateProvider.persist and load throw IllegalArgumentException for the new DateTime analyzer state types; incremental state persistence via saveStatesWith/aggregateWith will fail at runtime.
StateProvider.scala line 158-159 (persist):
case _ => throw new IllegalArgumentException(s"Unable to persist state for analyzer $analyzer.")and line 214-215 (load):case _ => throw new IllegalArgumentException(s"Unable to load state for analyzer $analyzer.")— no cases for MinDateTimeState, MaxDateTimeState, or DateTimeDistributionState.
Refutation trail (why this survived the Critic's disprove pass)
Hypothesis (Investigator): HdfsStateProvider.persist/load have no case for MinDateTimeState/MaxDateTimeState/DateTimeDistributionState, so using saveStatesWith/aggregateWith with a DateTime analyzer throws IllegalArgumentException.
Disprove attempt (Critic): Read StateProvider persist match (lines 84-159) and load match (lines 205-215): persist ends at ExactQuantile then throws at line 158-159; load ends at ExactQuantile then throws at line 214-215. No DateTime state cases in either. InMemoryStateProvider (used in tests) stores states in a map keyed by analyzer without a type match, so it is unaffected — but HdfsStateProvider throws.
The Critic's default verdict is OVERTURNED. UPHELD findings are those it tried — and failed — to refute.
| } | ||
| } | ||
|
|
||
| case class DateTimeMetric( |
There was a problem hiding this comment.
BUG: DateTimeMetric lacks a serialization case in MetricSerializer, causing FileSystemMetricsRepository and SparkTableMetricsRepository to throw IllegalArgumentException when saving successful MinimumDateTime or MaximumDateTime results.
AnalysisResultSerde.scala line 820-821:
case _ => throw new IllegalArgumentException(s"Unable to serialize metrics $metric.")— no DateTimeMetric case. InMemoryMetricsRepository (used in tests) bypasses serialization, which is why the added VerificationSuite test passes.
Refutation trail (why this survived the Critic's disprove pass)
Hypothesis (Investigator): DateTimeMetric has no case in AnalysisResultSerde.MetricSerializer, so persisting a successful MinimumDateTime/MaximumDateTime metric to a FileSystem/Spark repository throws IllegalArgumentException.
Disprove attempt (Critic): Read MetricSerializer (lines 764-826): cases exist for DoubleMetric, HistogramMetric, HistogramBinnedMetric, KeyedDoubleMetric, KLLMetric, then case _ => throw new IllegalArgumentException. No DateTimeMetric case. FileSystemMetricsRepository.save calls AnalysisResultSerde.serialize on successful metrics. DateTimeMetric with Success value would hit the fallthrough.
The Critic's default verdict is OVERTURNED. UPHELD findings are those it tried — and failed — to refute.
| override def metricValue(): Instant = maxValue | ||
| } | ||
|
|
||
| case class MaximumDateTime(column: String, where: Option[String] = None) |
There was a problem hiding this comment.
BUG: The three new DateTime analyzers (MaximumDateTime, MinimumDateTime, DateTimeDistribution) lack serialization cases in AnalyzerSerializer, causing repository-backed persistence to fail.
AnalysisResultSerde.scala line 470-471:
case _ => throw new IllegalArgumentException("Unable to serialize analyzer ...")— no cases for the new DateTime analyzers. AnalyzerContextSerializer calls serialize on each analyzer, so any FileSystem/Spark repository save will throw.
Refutation trail (why this survived the Critic's disprove pass)
Hypothesis (Investigator): AnalyzerSerializer has no case for MaximumDateTime/MinimumDateTime/DateTimeDistribution, so persisting these analyzers to a FileSystem/Spark repository throws IllegalArgumentException.
Disprove attempt (Critic): Read AnalyzerSerializer (lines 246-476): last case is KLLSketch, then case _ => throw new IllegalArgumentException("Unable to serialize analyzer ...") at line 470. No MaximumDateTime/MinimumDateTime/DateTimeDistribution cases. AnalyzerContextSerializer serializes each analyzer alongside its metric, so save fails on the analyzer regardless of metric type.
The Critic's default verdict is OVERTURNED. UPHELD findings are those it tried — and failed — to refute.
2 participants
Description of changes:
Add DateTimeMetric support. This chunk part of #299 to only DateTimeMetric support:
java.time.Instantisntead ofjava.sql.timestampBy submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.