add customsql analyzer

Author: wwitkowski

pydeequ/analyzers.py

@@ -360,6 +360,30 @@ def _analyzer_jvm(self):
         return self._deequAnalyzers.CountDistinct(to_scala_seq(self._jvm, self.columns))
 
 
+class CustomSql(_AnalyzerObject):
+    """
+    A custom SQL-based analyzer executing provided SQL expression.
+    The expression must return a single value.
+
+    :param str expression: A SQL expression to execute.
+    :param str where: A label used to distinguish this metric 
+        when running multiple custom SQL analyzers. Defaults to "*".
+    """
+
+    def __init__(self, expression: str, disambiguator: str = "*"):
+        self.expression = expression
+        self.disambiguator = disambiguator
+
+    @property
+    def _analyzer_jvm(self):
+        """
+        Returns the result of SQL expression execution.
+
+        :return self
+        """
+        return self._deequAnalyzers.CustomSql(self.expression, self.disambiguator)
+
+
 class DataType(_AnalyzerObject):
     """
     Data Type Analyzer. Returns the datatypes of column

tests/test_analyzers.py

@@ -14,6 +14,7 @@
     Compliance,
     Correlation,
     CountDistinct,
+    CustomSql,
     DataType,
     Distinctness,
     Entropy,
@@ -111,6 +112,14 @@ def CountDistinct(self, columns):
         df_from_json = self.spark.read.json(self.sc.parallelize([result_json]))
         self.assertEqual(df_from_json.select("value").collect(), result_df.select("value").collect())
         return result_df.select("value").collect()
+    
+    def CustomSql(self, expression, disambiguator="*"):
+        result = self.AnalysisRunner.onData(self.df).addAnalyzer(CustomSql(expression, disambiguator)).run()
+        result_df = AnalyzerContext.successMetricsAsDataFrame(self.spark, result)
+        result_json = AnalyzerContext.successMetricsAsJson(self.spark, result)
+        df_from_json = self.spark.read.json(self.sc.parallelize([result_json]))
+        self.assertEqual(df_from_json.select("value").collect(), result_df.select("value").collect())
+        return result_df.select("value").collect()
 
     def Datatype(self, column, where=None):
         result = self.AnalysisRunner.onData(self.df).addAnalyzer(DataType(column, where)).run()
@@ -298,6 +307,16 @@ def test_CountDistinct(self):
     def test_fail_CountDistinct(self):
         self.assertEqual(self.CountDistinct("b"), [Row(value=1.0)])
 
+    def test_CustomSql(self):
+        self.df.createOrReplaceTempView("input_table")
+        self.assertEqual(self.CustomSql("SELECT SUM(b) FROM input_table"), [Row(value=6.0)])
+        self.assertEqual(self.CustomSql("SELECT AVG(LENGTH(a)) FROM input_table"), [Row(value=3.0)])
+        self.assertEqual(self.CustomSql("SELECT MAX(c) FROM input_table"), [Row(value=6.0)])
+
+    @pytest.mark.xfail(reason="@unittest.expectedFailure")
+    def test_fail_CustomSql(self):
+        self.assertEqual(self.CustomSql("SELECT SUM(b) FROM input_table"), [Row(value=1.0)])
+
     def test_DataType(self):
         self.assertEqual(
             self.Datatype("b"),