Skip to content

Commit 577767b

Browse files
awang923anniezc
andauthored
added try_cast for min/max (#690)
Co-authored-by: anniezc <anniezc@amazon.com>
1 parent 0a31e51 commit 577767b

5 files changed

Lines changed: 38 additions & 2 deletions

File tree

src/main/scala/com/amazon/deequ/analyzers/Maximum.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ case class Maximum(column: String, where: Option[String] = None, analyzerOptions
4949
// The criterion returns a column where each row contains an array of 2 elements.
5050
// The first element of the array is a string that indicates if the row is "in scope" or "filtered" out.
5151
// The second element is the value used for calculating the metric. We use "element_at" to extract it.
52-
max(element_at(criterion, 2).cast(DoubleType)) :: Nil
52+
max(element_at(criterion, 2).try_cast(DoubleType)) :: Nil
5353
}
5454

5555
override def fromAggregationResult(result: Row, offset: Int): Option[MaxState] = {

src/main/scala/com/amazon/deequ/analyzers/Minimum.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ case class Minimum(column: String, where: Option[String] = None, analyzerOptions
4949
// The criterion returns a column where each row contains an array of 2 elements.
5050
// The first element of the array is a string that indicates if the row is "in scope" or "filtered" out.
5151
// The second element is the value used for calculating the metric. We use "element_at" to extract it.
52-
min(element_at(criterion, 2).cast(DoubleType)) :: Nil
52+
min(element_at(criterion, 2).try_cast(DoubleType)) :: Nil
5353
}
5454

5555
override def fromAggregationResult(result: Row, offset: Int): Option[MinState] = {

src/test/scala/com/amazon/deequ/analyzers/MaximumTest.scala

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,5 +59,16 @@ class MaximumTest extends AnyWordSpec with Matchers with SparkContextSpec with F
5959
.map(r => if (r == null) null else r.getAs[Double](tempColName))
6060
values shouldBe Seq(null, null, null, 5.0, 6.0, 7.0)
6161
}
62+
63+
"handle non-numeric strings" in withSparkSession { session =>
64+
val data = getDfWithNonNumericValues(session)
65+
66+
val att1Maximum = Maximum("att1")
67+
val state: Option[MaxState] = att1Maximum.computeStateFrom(data)
68+
val metric: DoubleMetric with FullColumn = att1Maximum.computeMetricFrom(state)
69+
70+
metric.value.isSuccess shouldBe true
71+
metric.value.get shouldBe 10.0
72+
}
6273
}
6374
}

src/test/scala/com/amazon/deequ/analyzers/MinimumTest.scala

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,5 +58,17 @@ class MinimumTest extends AnyWordSpec with Matchers with SparkContextSpec with F
5858
.map(r => if (r == null) null else r.getAs[Double](tempColName))
5959
values shouldBe Seq(null, null, null, 5.0, 6.0, 7.0)
6060
}
61+
62+
"handle non-numeric strings" in withSparkSession { session =>
63+
val data = getDfWithNonNumericValues(session)
64+
65+
val att1Minimum = Minimum("att1")
66+
val state: Option[MinState] = att1Minimum.computeStateFrom(data)
67+
val metric: DoubleMetric with FullColumn = att1Minimum.computeMetricFrom(state)
68+
69+
metric.value.isSuccess shouldBe true
70+
metric.value.get shouldBe 1.0
71+
}
72+
6173
}
6274
}

src/test/scala/com/amazon/deequ/utils/FixtureSupport.scala

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,19 @@ trait FixtureSupport {
260260
).toDF("item", "att1", "att2", "att3", "attNull")
261261
}
262262

263+
def getDfWithNonNumericValues(sparkSession: SparkSession): DataFrame = {
264+
import sparkSession.implicits._
265+
266+
Seq(
267+
("3", "foo", "10"),
268+
("10", "bar", "20"),
269+
("foo", "5", "baz"),
270+
("5", "10", "30"),
271+
("bar", "baz", "qux"),
272+
("1", "20", "40")
273+
).toDF("att1", "att2", "attAllNonNumeric")
274+
}
275+
263276
def getDfWithEscapeCharacters(sparkSession: SparkSession): DataFrame = {
264277
import sparkSession.implicits._
265278
// The names are with escape characters '

0 commit comments

Comments
 (0)