0.1 initial commit

Author: GeekSheikh

.gitignore

@@ -0,0 +1,12 @@
+/.idea/
+*.iml
+#local spark context data from unit tests
+spark-warehouse/
+
+#Build dirctory for maven/sbt
+target/
+project/project/
+project/target/
+*.DS_Store
+/target/
+/project/

build.sbt

@@ -0,0 +1,18 @@
+name := "Rules_Engine"
+
+organization := "com.databricks"
+
+version := "0.1"
+
+scalaVersion := "2.11.12"
+scalacOptions ++= Seq("-Xmax-classfile-name", "78")
+
+libraryDependencies += "org.apache.spark" %% "spark-core" % "2.4.0"
+libraryDependencies += "org.apache.spark" %% "spark-mllib" % "2.4.0"
+libraryDependencies += "org.apache.spark" %% "spark-sql" % "2.4.0"
+
+lazy val commonSettings = Seq(
+  version := "0.1",
+  organization := "com.databricks",
+  scalaVersion := "2.11.12"
+)

src/main/scala/com/databricks/labs/validation/Example.scala

@@ -0,0 +1,79 @@
+package com.databricks.labs.validation
+
+import com.databricks.labs.validation.utils.{SparkSessionWrapper, MinMaxFunc}
+import com.databricks.labs.validation.utils.Structures._
+import org.apache.spark.sql.{Column, functions}
+import org.apache.spark.sql.functions.col
+
+object Example extends App with SparkSessionWrapper {
+  import spark.implicits._
+
+  /**
+   * Validation example
+   * Passing pre-built array of rules into a RuleSet and validating a non-grouped dataframe
+   */
+
+  /**
+   * Example of a proper UDF to simplify rules logic. Simplification UDFs should take in zero or many
+   * columns and return one column
+   * @param retailPrice column 1
+   * @param scanPrince column 2
+   * @return result column of applied logic
+   */
+  def getDiscountPercentage(retailPrice: Column, scanPrince: Column): Column = {
+    ((retailPrice - scanPrince) / retailPrice).alias("total_discount_percent")
+  }
+
+  // Example of creating array of custom rules
+  val specializedRules = Array(
+    Rule("Reasonable_sku_counts", col("sku"), functions.count, "sku_count", Bounds(lower = 20.0, upper = 200.0)),
+    Rule("Max_allowed_discount", getDiscountPercentage(col("retail_price"), col("scan_price")),
+      functions.max, "total_discount", Bounds(upper = 90.0))
+  )
+
+  // It's common to generate many mins and max boundaries. These can be generated easily
+  // The generator function can easily be extended or overridden to satisfy more complex requirements
+  val minMaxPriceDefs = Array(
+    MinMaxRuleDef("MinMax_Sku_Price", col("retail_price"), Bounds(0.0, 350.0)),
+    MinMaxRuleDef("MinMax_Sku_Price", col("retail_price"), Bounds(0.0, 350.0)),
+    MinMaxRuleDef("MinMax_Sku_Price", col("retail_price"), Bounds(0.0, 350.0))
+
+  )
+  val minMaxPriceRules = RuleSet.generateMinMaxRules(minMaxPriceDefs: _*)
+
+  val x = Array(("MinMax_Sku_Price", col("retail_price"), Bounds(0.0, 350.0)))
+
+  // Generate a single stand alone rule to demonstrate builder pattern below
+  val standAloneRule = Rule("stddev_retail_price", col("retail_price"), functions.stddev, "stddev_price", Bounds(upper = 13.0))
+
+  //TODO - validate datetime
+  // TODO - validate distinct keys
+  // Test, example data frame
+  val df = sc.parallelize(Seq(
+    (1001, 123456, 9.32, 8.99, 4.23),
+    (1001, 123256, 19.99, 16.49, 12.99),
+    (1001, 123456, 0.99, 0.99, 0.10),
+    (1001, 123456, 0.98, 0.90, 0.10), // non_distinct sku
+    (1002, 122987, 9.99, 9.49, 6.49),
+    (1002, 173544, 1.29, 0.99, 1.23),
+    (1002, 168212, 3.29, 1.99, 1.23),
+    (1002, 365423, 1.29, 0.99, 1.23),
+    (1002, 3897615, 14.99, 129.99, 1.23),
+    (1003, 163212, 3.29, 1.99, 1.23)
+  )).toDF("region", "store_id", "sku", "retail_price", "scan_price", "cost")
+
+  // Doing the validation
+  // The validate method will return the rules report dataframe which breaks down which rules passed and which
+  // rules failed and how/why. The second return value returns a boolean to determine whether or not all tests passed
+  val (rulesReport, passed) = RuleSet(df)
+    .add(specializedRules)
+    .add(standAloneRule)
+    .add(minMaxPriceRules)
+    .validate
+
+  rulesReport.show(false)
+
+
+
+
+}

src/main/scala/com/databricks/labs/validation/Rule.scala

@@ -0,0 +1,152 @@
+package com.databricks.labs.validation
+
+import com.databricks.labs.validation.utils.MinMaxFunc
+import com.databricks.labs.validation.utils.Structures.{Bounds, ValidNumerics, ValidStrings}
+import org.apache.spark.sql.Column
+
+class Rule {
+
+  private var _ruleName: String = _
+  private var _column: Column = _
+  private var _aggFunc: Column => Column = _
+  private var _alias: String = _
+  private var _boundaries: Bounds = _
+  private var _validNumerics: ValidNumerics = _
+  private var _validStrings: ValidStrings = _
+  private var _dateTimeLogic: Column = _
+  private var _ruleType: String = _
+  private var _by: Seq[Column] = _
+
+  private def setRuleName(value: String): this.type = {
+    _ruleName = value
+    this
+  }
+  private def setColumn(value: Column): this.type = {
+    _column = value
+    this
+  }
+  private def setAggFunc(value: Column => Column): this.type = {
+    _aggFunc = value
+    this
+  }
+  private def setAlias(value: String): this.type = {
+    _alias = value
+    this
+  }
+  private def setBoundaries(value: Bounds): this.type = {
+    _boundaries = value
+    this
+  }
+  private def setByCols(value: Seq[Column]): this.type = {
+    _by = value
+    this
+  }
+  private def setValidNumerics(value: ValidNumerics): this.type = {
+    _validNumerics = value
+    this
+  }
+  private def setValidStrings(value: ValidStrings): this.type = {
+    _validStrings = value
+    this
+  }
+  private def setDateTimeLogic(value: Column): this.type = {
+    _dateTimeLogic = value
+    this
+  }
+  private def setRuleType(value: String): this.type = {
+    _ruleType = value
+    this
+  }
+
+  def ruleName: String = _ruleName
+  def inputColumn: Column = _column
+  def aggFunc: Column => Column = _aggFunc //TODO - add .toString
+  def alias: String = _alias
+  def boundaries: Bounds = _boundaries
+  def groupByColumns: Seq[Column] = _by //TODO - add .toString
+  def validNumerics: ValidNumerics = _validNumerics
+  def validStrings: ValidStrings = _validStrings
+  def dateTimeLogic: Column = _dateTimeLogic
+  def ruleType: String = _ruleType
+
+}
+
+object Rule {
+
+  def apply(
+           ruleName: String,
+           column: Column,
+           aggFunc: Column => Column,
+           alias: String,
+           boundaries: Bounds,
+           by: Column*
+           ): Rule = {
+
+    new Rule()
+      .setRuleName(ruleName)
+      .setColumn(column)
+      .setAggFunc(aggFunc)
+      .setAlias(alias)
+      .setBoundaries(boundaries)
+      .setRuleType("bounds")
+      .setByCols(by)
+  }
+
+  def apply(
+             ruleName: String,
+             column: Column,
+             aggFunc: Column => Column, // TODO -- Handle aggs
+             alias: String,
+             validNumerics: ValidNumerics,
+             by: Column*
+           ): Rule = {
+
+    new Rule()
+      .setRuleName(ruleName)
+      .setColumn(column)
+      .setAggFunc(aggFunc)
+      .setAlias(alias)
+      .setValidNumerics(validNumerics)
+      .setRuleType("validNumerics")
+      .setByCols(by)
+  }
+
+  def apply(
+    ruleName: String,
+    column: Column,
+    aggFunc: Column => Column, // TODO - Handle Aggs
+    alias: String,
+    validStrings: ValidStrings,
+    by: Column*
+  ): Rule = {
+
+    new Rule()
+      .setRuleName(ruleName)
+      .setColumn(column)
+      .setAggFunc(aggFunc)
+      .setAlias(alias)
+      .setValidStrings(validStrings)
+      .setRuleType("validStrings")
+      .setByCols(by)
+  }
+
+  def apply(
+             ruleName: String,
+             column: Column,
+             aggFunc: Column => Column, // TODO - handle aggs
+             alias: String,
+             dateTimeLogic: Column,
+             by: Column*
+           ): Rule = {
+
+    new Rule()
+      .setRuleName(ruleName)
+      .setColumn(column)
+      .setAggFunc(aggFunc)
+      .setAlias(alias)
+      .setDateTimeLogic(dateTimeLogic)
+      .setRuleType("dateTime")
+      .setByCols(by)
+  }
+
+}