Adds support for UniqueCheck (#11)

dougb · web-flow · commit 102cfb9791b3 · 2019-06-11T16:38:55.000-04:00
* Adds support for UniqueCheck

* Changes from code review feedback.

* Updates package and some imports.
diff --git a/README.md b/README.md
@@ -200,7 +200,7 @@ To validate an `.orc` file, specify `orcFile` and the path to the file, see belo
 
 ### Validators
 
- The third section are the validators. Currently 5 validators are supported `columnMaxCheck`, `negativeCheck`, `nullCheck`, `rangeCheck` and `rowCount`. To specify a validator, you first specify the `type` as one of the validators, then specify the arguments for that validator.
+  The third section are the validators. To specify a validator, you first specify the type as one of the validators, then specify the arguments for that validator. Currently supported validators are listed below:
 
 #### `columnMaxCheck`
 
@@ -250,6 +250,15 @@ The minimum number of rows a table must have to pass the validator.
 
 See Example Config file below to see how the checks are configured.
 
+#### `uniqueCheck`
+
+This check is used to make sure all rows in the table are unique, only the columns specified are used to determine uniqueness.
+This is a costly check and requires an additional pass through the table.
+
+| Arg | Type | Description |
+|-----|------|-------------|
+| `columns` | Array[String] | Each set of values in these columns must be unique.
+
 ## Example Config
 
 ```yaml
diff --git a/src/main/scala/com/target/data_validator/Main.scala b/src/main/scala/com/target/data_validator/Main.scala
@@ -75,7 +75,10 @@ object Main extends LazyLogging with EventLog {
     mainConfig: CmdLineOptions,
     config: ValidatorConfig,
     varSub: VarSubstitution
-  ): Boolean = config.quickChecks(spark, varSub)
+  ): Boolean = {
+    logger.info("Running sparkChecks")
+    Seq(config.quickChecks(spark, varSub), config.costlyChecks(spark, varSub)).exists(x => x)
+  }
 
   /*
     * There are 2 types of errors we return (fatal, validator_status)
diff --git a/src/main/scala/com/target/data_validator/ValidatorConfig.scala b/src/main/scala/com/target/data_validator/ValidatorConfig.scala
@@ -47,9 +47,15 @@ case class ValidatorConfig(
   }
 
   def quickChecks(session: SparkSession, dict: VarSubstitution): Boolean = {
+    logger.info("Running Quick Checks...")
     tables.map(_.quickChecks(session, dict)(this)).exists(x => x)
   }
 
+  def costlyChecks(session: SparkSession, dict: VarSubstitution): Boolean = {
+    logger.info("Running Costly Checks...")
+    tables.map(_.costlyChecks(session, dict)(this)).exists(x => x)
+  }
+
   def generateHTMLReport(): Tag = html(h1("Validator Report"), hr(), tables.map(_.generateHTMLReport()))
 
   def substituteVariables(varSub: VarSubstitution): Option[ValidatorConfig] = {
diff --git a/src/main/scala/com/target/data_validator/ValidatorTable.scala b/src/main/scala/com/target/data_validator/ValidatorTable.scala
@@ -1,6 +1,6 @@
 package com.target.data_validator
 
-import com.target.data_validator.validator.{ColumnBased, RowBased, ValidatorBase}
+import com.target.data_validator.validator.{CheapCheck, ColumnBased, CostlyCheck, RowBased, ValidatorBase}
 import org.apache.spark.sql.{Column, DataFrame, Row, SparkSession}
 import org.apache.spark.sql.catalyst.expressions.{Alias, Expression}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{Count, Sum}
@@ -75,9 +75,13 @@ abstract class ValidatorTable(
 
   def quickChecks(session: SparkSession, dict: VarSubstitution)(implicit vc: ValidatorConfig): Boolean = {
     val dataFrame = open(session).get
-    val checkSelects: Seq[Expression] = checks.map {
+    val qc: List[CheapCheck] = checks.flatMap {
+      case cc: CheapCheck => Some(cc)
+      case _ => None
+    }
+    val checkSelects: Seq[Expression] = qc.map {
       case colChk: ColumnBased => colChk.select(dataFrame.schema, dict)
-      case chk => Sum(chk.select(dataFrame.schema, dict)).toAggregateExpression()
+      case chk: RowBased => Sum(chk.select(dataFrame.schema, dict)).toAggregateExpression()
     }
 
     val cols: Seq[Column] = createCountSelect() ++ checkSelects.zipWithIndex.map {
@@ -98,8 +102,8 @@ abstract class ValidatorTable(
     logger.info(s"Total Rows Processed: $count")
     addEvent(ValidatorCounter(s"RowCount for $label", count))
 
-    val failed = checks.zipWithIndex.map {
-      case (check: ValidatorBase, idx: Int) => check.quickCheck(results, count, idx + 1)
+    val failed = qc.zipWithIndex.map {
+      case (check: CheapCheck, idx: Int) => check.quickCheck(results, count, idx + 1)
     }.exists(x => x)
 
     if (failed) {
@@ -112,6 +116,15 @@ abstract class ValidatorTable(
     failed
   }
 
+  def costlyChecks(session: SparkSession, dict: VarSubstitution)(implicit  vc: ValidatorConfig): Boolean = {
+    val df = open(session).get
+    val cc = checks.flatMap {
+      case cc: CostlyCheck => Some(cc)
+      case _ => None
+    }
+    cc.map(_.costlyCheck(df)).exists(x => x)
+  }
+
   def quickErrorDetails(dataFrame: DataFrame, dict: VarSubstitution)(implicit vc: ValidatorConfig): Unit = {
     val keySelect = createKeySelect(dataFrame)
     val failedChecksWithIndex = checks
diff --git a/src/main/scala/com/target/data_validator/validator/ColumnBased.scala b/src/main/scala/com/target/data_validator/validator/ColumnBased.scala
@@ -10,7 +10,7 @@ import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.catalyst.expressions.aggregate.Max
 import org.apache.spark.sql.types._
 
-abstract class ColumnBased(column: String, condTest: Expression) extends ValidatorBase {
+abstract class ColumnBased(column: String, condTest: Expression) extends CheapCheck {
   override def select(schema: StructType, dict: VarSubstitution): Expression = condTest
 
   // ColumnBased checks don't have per row error details.
diff --git a/src/main/scala/com/target/data_validator/validator/JsonDecoders.scala b/src/main/scala/com/target/data_validator/validator/JsonDecoders.scala
@@ -14,6 +14,7 @@ object JsonDecoders extends LazyLogging {
       case "negativeCheck" => c.as[NegativeCheck]
       case "columnMaxCheck" => c.as[ColumnMaxCheck]
       case "rangeCheck" => RangeCheck.fromJson(c)
+      case "uniqueCheck" => UniqueCheck.fromJson(c)
       case x => logger.error(s"Unknown Check `$x` in config!")
         throw new RuntimeException(s"Unknown Check in config `$x`")
     }
diff --git a/src/main/scala/com/target/data_validator/validator/RowBased.scala b/src/main/scala/com/target/data_validator/validator/RowBased.scala
@@ -10,7 +10,7 @@ import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.types.{NumericType, StructType}
 
-abstract class RowBased extends ValidatorBase {
+abstract class RowBased extends CheapCheck {
 
   val column: String
 
diff --git a/src/main/scala/com/target/data_validator/validator/UniqueCheck.scala b/src/main/scala/com/target/data_validator/validator/UniqueCheck.scala
@@ -0,0 +1,58 @@
+package com.target.data_validator.validator
+
+import com.target.data_validator.{ValidatorError, ValidatorGood, ValidatorTimer, VarSubstitution}
+import com.typesafe.scalalogging.LazyLogging
+import io.circe.{DecodingFailure, HCursor, Json}
+import io.circe.syntax._
+import org.apache.spark.sql.{Column, DataFrame}
+
+case class UniqueCheck(columns: Seq[String]) extends CostlyCheck {
+
+  override def substituteVariables(dict: VarSubstitution): ValidatorBase = {
+    val newColumns = columns.map(getVarSub(_, "columns", dict))
+    val ret = UniqueCheck(newColumns)
+    this.getEvents.foreach(ret.addEvent)
+    ret
+  }
+
+  override def configCheck(df: DataFrame): Boolean = {
+    columns.exists(findColumnInDataFrame(df, _).isEmpty)
+  }
+
+  override def toJson: Json = {
+    import com.target.data_validator.JsonEncoders.eventEncoder
+    val fields = Seq(
+      ("type", Json.fromString("uniqueCheck")),
+      ("columns", Json.fromValues(columns.map(Json.fromString))),
+      ("failed", Json.fromBoolean(failed)),
+        ("events", this.getEvents.asJson))
+
+    Json.fromFields(fields)
+  }
+
+  override def costlyCheck(df: DataFrame): Boolean = {
+    val cols = columns.map(new Column(_))
+    val timer = new ValidatorTimer(s"UniqueCheck($columns)")
+    addEvent(timer)
+    // Note: this computes the count of the number of distinct keys (if you will) that have at least one duplicated row.
+    // It's not number of duplicated rows.
+    val ret = timer.time(df.select(cols: _*).groupBy(cols: _*).count().where("count > 1").count())
+    logger.info(s"costlyCheck: cols:$cols ret:$ret")
+    if (ret > 0) {
+      addEvent(ValidatorError(s"$ret duplicates found!"))
+    } else {
+      addEvent(ValidatorGood("no duplicates found."))
+    }
+
+    failed
+  }
+}
+
+object UniqueCheck extends LazyLogging {
+
+  def fromJson(c: HCursor): Either[DecodingFailure, ValidatorBase] = {
+    val columns = c.downField("columns").as[Seq[String]]
+    columns.right.map(UniqueCheck(_))
+  }
+
+}
diff --git a/src/main/scala/com/target/data_validator/validator/ValidatorBase.scala b/src/main/scala/com/target/data_validator/validator/ValidatorBase.scala
@@ -23,10 +23,6 @@ abstract class ValidatorBase(
 
   def configCheck(df: DataFrame): Boolean
 
-  def select(schema: StructType, dict: VarSubstitution): Expression
-
-  def quickCheck(r: Row, count: Long, idx: Int): Boolean
-
   def generateHTMLReport: Tag = {
     val d = div(cls := "check_report")
     if (failed) {
@@ -64,7 +60,7 @@ abstract class ValidatorBase(
         addEvent(ValidatorError(msg))
       }
     } else {
-      val msg = s"Column: $column not found in table."
+      val msg = s"Column: '$column' not found in table."
       logger.error(msg)
       addEvent(ValidatorError(msg))
     }
@@ -74,7 +70,9 @@ abstract class ValidatorBase(
   private[validator] def findColumnInDataFrame(dataFrame: DataFrame, column: String): Option[StructField] = {
     val ret = dataFrame.schema.fields.find(_.name == column)
     if (ret.isEmpty) {
-      addEvent(ValidatorError(s"Column: $column not found in schema."))
+      val msg = s"Column: '$column' not found in schema."
+      logger.error(msg)
+      addEvent(ValidatorError(msg))
     }
     ret
   }
@@ -244,3 +242,19 @@ object ValidatorBase extends LazyLogging {
     ret
   }
 }
+
+/**
+* CheapChecks are checks that can be combined into the same pass through a table.
+*/
+trait CheapCheck extends ValidatorBase {
+  def select(schema: StructType, dict: VarSubstitution): Expression
+
+  def quickCheck(r: Row, count: Long, idx: Int): Boolean
+}
+
+/**
+* CostlyChecks are checks that require their own pass through the table and therefore are most costly.
+*/
+trait CostlyCheck extends ValidatorBase {
+  def costlyCheck(df: DataFrame): Boolean
+}
diff --git a/src/test/scala/com/target/data_validator/validator/RangeCheckSpec.scala b/src/test/scala/com/target/data_validator/validator/RangeCheckSpec.scala
@@ -64,7 +64,7 @@ class RangeCheckSpec extends FunSpec with Matchers with TestingSparkSession {
           None
         )
         assert(sut.configCheck(df))
-        assert(sut.getEvents contains ValidatorError("Column: bad_column_name not found in schema."))
+        assert(sut.getEvents contains ValidatorError("Column: 'bad_column_name' not found in schema."))
         assert(sut.failed)
       }
 
diff --git a/src/test/scala/com/target/data_validator/validator/UniqueCheckSpec.scala b/src/test/scala/com/target/data_validator/validator/UniqueCheckSpec.scala
@@ -0,0 +1,132 @@
+package com.target.data_validator.validator
+
+import com.target.TestingSparkSession
+import com.target.data_validator._
+import io.circe.Json
+import org.apache.spark.sql.{DataFrame, Row, SparkSession}
+import org.apache.spark.sql.types._
+import org.scalatest.{FunSpec, Matchers}
+
+class UniqueCheckSpec extends FunSpec with Matchers with TestingSparkSession {
+
+  val schema = StructType(List(StructField("item", StringType),
+    StructField("location", IntegerType),
+    StructField("price", DoubleType)))
+
+  val defData = List(Row("Eggs", 1, 4.00), Row("Milk", 1, 10.27),
+    Row("Eggs", 1, 5.00), Row("Eggs", 2, 2.00))
+  def mkDataFrame(spark: SparkSession, data: List[Row]): DataFrame = spark.createDataFrame(sc.parallelize(data), schema)
+
+  describe("fromJson") {
+    it("create fromJson") {
+      import com.target.data_validator.validator.JsonDecoders.decodeChecks
+      val yaml =
+        """---
+          |- type: uniqueCheck
+          |  columns:
+          |   - foo
+          |   - bar
+        """.stripMargin
+      val json = io.circe.yaml.parser.parse(yaml).right.getOrElse(Json.Null)
+      val sut = json.as[Array[ValidatorBase]]
+      assert(sut.isRight)
+      assert(sut.right.get contains UniqueCheck(Array("foo", "bar")))
+    }
+  }
+
+  describe ("substituteVariables") {
+    it("replaces variables") {
+      val dict = new VarSubstitution
+      dict.addString("col1", "foo")
+      dict.addString("col2", "bar")
+      val sut = UniqueCheck(List("${col1}", "$col2"))
+      assert(sut.substituteVariables(dict) == UniqueCheck(List("foo", "bar")))
+      assert(!sut.failed)
+    }
+
+  }
+
+  describe ("configCheck") {
+    it("good columns") {
+      val sut = UniqueCheck(List("item", "location"))
+      val df = mkDataFrame(spark, defData)
+      assert(!sut.configCheck(df))
+      assert(!sut.failed)
+    }
+
+    it("bad column") {
+      val sut = UniqueCheck(List("item", "city"))
+      val df = mkDataFrame(spark, defData)
+      assert(sut.configCheck(df))
+      assert(sut.failed)
+    }
+
+  }
+
+  describe("costlyCheck") {
+
+    it("finds error") {
+      val sut = UniqueCheck(Seq("item"))
+      val df = mkDataFrame(spark, defData)
+      assert(sut.costlyCheck(df))
+      assert(sut.failed)
+      assert(sut.getEvents contains ValidatorError("1 duplicates found!"))
+    }
+
+    it("finds error with multiple columns") {
+      val sut = UniqueCheck(Seq("item", "location"))
+      val df = mkDataFrame(spark, defData)
+      assert(sut.costlyCheck(df))
+      assert(sut.failed)
+      assert(sut.getEvents contains ValidatorError("1 duplicates found!"))
+    }
+
+    it("no error") {
+      val sut = UniqueCheck(Seq("price"))
+      val df = mkDataFrame(spark, defData)
+      assert(!sut.costlyCheck(df))
+      assert(!sut.failed)
+      assert(sut.getEvents contains ValidatorGood("no duplicates found."))
+    }
+  }
+
+  describe("toJson") {
+
+    it("generates correct json") {
+      val sut = UniqueCheck(Seq("item"))
+      assert(sut.toJson == Json.fromFields(Seq(
+        ("type", Json.fromString("uniqueCheck")),
+        ("columns", Json.fromValues(List(Json.fromString("item")))),
+        ("failed", Json.fromBoolean(false)),
+        ("events", Json.fromValues(Seq.empty)))))
+    }
+  }
+
+  describe("completeExample") {
+    it("happy path that finds error") {
+      val uc = UniqueCheck(List("item"))
+      val dict = new VarSubstitution
+      val df = mkDataFrame(spark, defData)
+      val sut = ValidatorConfig(1, 1, None, detailedErrors = false, None, None,
+        List(ValidatorDataFrame(df, None, None, List(uc))))
+
+      assert(!sut.configCheck(spark, dict))
+      assert(!sut.quickChecks(spark, dict))
+      assert(sut.costlyChecks(spark, dict))
+      assert(sut.failed)
+    }
+
+    it("happy path that doesn't find error") {
+      val uc = UniqueCheck(List("price"))
+      val dict = new VarSubstitution
+      val df = mkDataFrame(spark, defData)
+      val sut = ValidatorConfig(1, 1, None, detailedErrors = false, None, None,
+        List(ValidatorDataFrame(df, None, None, List(uc))))
+
+      assert(!sut.configCheck(spark, dict))
+      assert(!sut.costlyChecks(spark, dict))
+      assert(!sut.failed)
+    }
+  }
+
+}

Original file line number	Diff line number	Diff line change
`@@ -47,9 +47,15 @@ case class ValidatorConfig(`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`def quickChecks(session: SparkSession, dict: VarSubstitution): Boolean = {`
	`50`	`+ logger.info("Running Quick Checks...")`
`50`	`51`	`tables.map(_.quickChecks(session, dict)(this)).exists(x => x)`
`51`	`52`	`}`
`52`	`53`
	`54`	`+ def costlyChecks(session: SparkSession, dict: VarSubstitution): Boolean = {`
	`55`	`+ logger.info("Running Costly Checks...")`
	`56`	`+ tables.map(_.costlyChecks(session, dict)(this)).exists(x => x)`
	`57`	`+ }`
	`58`	`+`
`53`	`59`	`def generateHTMLReport(): Tag = html(h1("Validator Report"), hr(), tables.map(_.generateHTMLReport()))`
`54`	`60`
`55`	`61`	`def substituteVariables(varSub: VarSubstitution): Option[ValidatorConfig] = {`
Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,7 @@ object JsonDecoders extends LazyLogging {`
`14`	`14`	`case "negativeCheck" => c.as[NegativeCheck]`
`15`	`15`	`case "columnMaxCheck" => c.as[ColumnMaxCheck]`
`16`	`16`	`case "rangeCheck" => RangeCheck.fromJson(c)`
	`17`	`+ case "uniqueCheck" => UniqueCheck.fromJson(c)`
`17`	`18`	case x => logger.error(s"Unknown Check `$x` in config!")
`18`	`19`	throw new RuntimeException(s"Unknown Check in config `$x`")
`19`	`20`	`}`
Original file line number	Diff line number	Diff line change
`@@ -64,7 +64,7 @@ class RangeCheckSpec extends FunSpec with Matchers with TestingSparkSession {`
`64`	`64`	`None`
`65`	`65`	`)`
`66`	`66`	`assert(sut.configCheck(df))`
`67`		`- assert(sut.getEvents contains ValidatorError("Column: bad_column_name not found in schema."))`
	`67`	`+ assert(sut.getEvents contains ValidatorError("Column: 'bad_column_name' not found in schema."))`
`68`	`68`	`assert(sut.failed)`
`69`	`69`	`}`
`70`	`70`