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.
| override def zero: Map[Long, Long] = Map.empty[Long, Long] | ||
|
|
||
| override def reduce(agg: Map[Long, Long], input: Instant): Map[Long, Long] = { | ||
| val dateTime = input.toEpochMilli |
There was a problem hiding this comment.
Null safety: if the column contains null values, input will be null and input.toEpochMilli will throw a NullPointerException. You should add a null check and skip null inputs (return agg unchanged).
|
|
||
| // Define encoder for output | ||
| def outputEncoder: Encoder[Map[Long, Long]] = ExpressionEncoder() | ||
| } No newline at end of file |
There was a problem hiding this comment.
Missing newline at end of file.
| 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. |
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.