tutorials/test_data_quality_at_scale.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Testing data quality at scale with PyDeequ\n",
    "\n",
    "Authors: Vitalina Komashko (komashk@), Calvin Wang (calviwan@), Chris Ghyzel (cghyzel@), Joan Aoanan (jaoanan@), Veronika Megler (meglerv@) \n",
    "\n",
    "__Updated June 2024 to use a new dataset, added additional library usage examples.__\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook accompanies AWS Blog post [Testing data quality at scale with PyDeequ](https://aws.amazon.com/blogs/big-data/testing-data-quality-at-scale-with-pydeequ/).\n",
    "\n",
    "You generally write unit tests for your code, but do you also test your data? Incoming data quality can make or break your application. Incorrect, missing, or malformed data can have a large impact on production systems. Examples of data quality issues include the following:\n",
    "\n",
    "- Missing values can lead to failures in the production system that require non-null values (`NullPointerException`)\n",
    "- Changes in the distribution of data can lead to unexpected outputs of machine learning (ML) models\n",
    "- Aggregations of incorrect data can lead to misguided business decisions\n",
    "\n",
    "In this post, we introduce PyDeequ, an open source Python wrapper over [Deequ](https://aws.amazon.com/blogs/big-data/test-data-quality-at-scale-with-deequ/) (an open source tool developed and used at Amazon). Deequ is written in [Scala](https://www.scala-lang.org/), whereas PyDeequ allows you to use its data quality and testing capabilities from Python and PySpark, the language of choice for many data scientists. PyDeequ democratizes and extends the power of Deequ by allowing you to use it alongside the many data science libraries that are available in that language. Furthermore, PyDeequ allows for fluid interface with [pandas](https://pandas.pydata.org/) DataFrames as opposed to restricting within [Apache Spark](https://spark.apache.org/) DataFrames.\n",
    "\n",
    "Deequ allows you to calculate data quality metrics for your dataset, define and verify data quality constraints, and be informed about changes in data distribution. Instead of implementing checks and verification algorithms on your own, you can focus on describing how your data should look. Deequ supports you by suggesting checks for you. Deequ is implemented on top of Apache Spark and is designed to scale with large datasets (billions of rows) that typically live in a data lake, distributed file system, or a data warehouse. PyDeequ gives you access to this capability, but also allows you to use it from the familiar environment of your Python [Jupyter](https://jupyter.org/) notebook.\n",
    "\n",
    "## Deequ at Amazon \n",
    "\n",
    "Deequ is used internally at Amazon to verify the quality of many large production datasets. Dataset producers can add and edit data quality constraints. The system computes data quality metrics on a regular basis (with every new version of a dataset), verifies constraints defined by dataset producers, and publishes datasets to consumers in case of success. In error cases, dataset publication can be stopped, and producers are notified to take action. Data quality issues don’t propagate to consumer data pipelines, reducing their area of impact.\n",
    "\n",
    "Deequ is also used within [Amazon SageMaker Model Monitor](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor.html#model-monitor-how-it-works). Now with the availability of PyDeequ, you can use it from a broader set of environments — [Amazon SageMaker](https://aws.amazon.com/sagemaker/), [AWS Glue](https://aws.amazon.com/glue/), [Amazon EMR](https://aws.amazon.com/emr/), and more.\n",
    "\n",
    "## Overview of PyDeequ\n",
    "\n",
    "Let’s look at PyDeequ’s main components, and how they relate to Deequ (shown in the following diagram). \n",
    "\n",
    "- __Metrics computation__ – Deequ computes data quality metrics, which are statistics such as completeness, maximum, or correlation. Deequ uses Spark to read from sources such as [Amazon Simple Storage Service](https://aws.amazon.com/s3/) (Amazon S3) and compute metrics through an optimized set of aggregation queries. You have direct access to the raw metrics computed on the data.\n",
    "- __Constraint verification__ – As a user, you focus on defining a set of data quality constraints to be verified. Deequ takes care of deriving the required set of metrics to be computed on the data. Deequ generates a data quality report, which contains the result of the constraint verification.\n",
    "- __Constraint suggestion__ – You can choose to define your own custom data quality constraints or use the automated constraint suggestion methods that profile the data to infer useful constraints.\n",
    "- __Python wrappers__ – You can call each Deequ function using Python syntax. The wrappers translate the commands to the underlying Deequ calls and return their response.\n",
    "\n",
    "![pydeequ-spark-components](../imgs/pydeequ_architecture.jpg)\n",
    "\n",
    "**Figure 1. Overview of PyDeequ components.** \n",
    "\n",
    "## Solution overview \n",
    "\n",
    "As a running example, we have generated a synthetic reviews dataset and introduced various data issues. We demonstrate how to detect these issues using PyDeequ. We begin the way many data science projects do: with initial data exploration and assessment in a Jupyter notebook.\n",
    "\n",
    "During the data exploration phase, we want to answer some basic questions about the data:\n",
    "\n",
    "- Are there fields that have missing values?\n",
    "- How many distinct categories are there in the categorical fields?\n",
    "- Are there correlations between some key features?\n",
    "- If there are two supposedly similar datasets (such as different categories or different time periods), are they really similar?\n",
    "- We also show you how to scale this approach to large-scale datasets, using the same code on an EMR cluster. This is how you’d likely do your ML training as you move into a production setting.\n",
    "\n",
    "\n",
    "\n",
    "## Setup\n",
    "\n",
    "In this section we will show how to set up PyDeequ in [SageMaker Notebooks](https://docs.aws.amazon.com/sagemaker/latest/dg/nbi.html).\n",
    "\n",
    "We use the default VPC for SageMaker Notebooks. The examples presented here use PyDeequ library version 1.2.0 (latest at the time of the update to this notebook) and tested in a SageMaker Notebook instance ml.m5.2xlarge, `conda_python3` kernel.\n",
    " \n",
    "1. Create a new notebook instance. \n",
    "\n",
    "As of version 1.1.0, PyDeequ supports Spark up to version 3.3.0. Your PyDeequ version has to work with your version of Spark.\n",
    "\n",
    "2. In the notebook, run the following lines in a code cell to specify `SPARK_VERSION`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ[\"SPARK_VERSION\"] = '3.3' "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "3. Install PyDeequ module. For consistency, we'll set the PyDeequ version too."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pydeequ==1.2.0 in /home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages (1.2.0)\n",
      "Requirement already satisfied: numpy>=1.14.1 in /home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages (from pydeequ==1.2.0) (1.22.4)\n",
      "Requirement already satisfied: pandas>=0.23.0 in /home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages (from pydeequ==1.2.0) (2.2.1)\n",
      "Requirement already satisfied: python-dateutil>=2.8.2 in /home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages (from pandas>=0.23.0->pydeequ==1.2.0) (2.9.0)\n",
      "Requirement already satisfied: pytz>=2020.1 in /home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages (from pandas>=0.23.0->pydeequ==1.2.0) (2024.1)\n",
      "Requirement already satisfied: tzdata>=2022.7 in /home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages (from pandas>=0.23.0->pydeequ==1.2.0) (2024.1)\n",
      "Requirement already satisfied: six>=1.5 in /home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages (from python-dateutil>=2.8.2->pandas>=0.23.0->pydeequ==1.2.0) (1.16.0)\n",
      "Note: you may need to restart the kernel to use updated packages.\n"
     ]
    }
   ],
   "source": [
    "pip install pydeequ==1.2.0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "4. To import the modules, run the following commands in a code cell:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import sagemaker_pyspark\n",
    "import pydeequ"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This completes the steps specific to SageMaker Notebooks."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Start a PySpark Session\n",
    "\n",
    "In the cell below we import modules and set up a Spark session with the following configurations:\n",
    "\n",
    "- `config(\"spark.driver.extraClassPath\", classpath)` to prepend extra classpath entries to the classpath of the driver\n",
    "- `config(\"spark.jars.packages\", pydeequ.deequ_maven_coord)` to provide Maven of jars to include on the driver and executor classpaths\n",
    "- `config(\"spark.jars.excludes\", pydeequ.f2j_maven_coord` to exclude jars to avoid conflicts\n",
    "- `config(\"spark.driver.memory\", \"15g\")` to increase Java heap space\n",
    "- `config(\"spark.sql.parquet.int96RebaseModeInRead\", \"CORRECTED\")` to read the datetime values as is. In our synthetic dataset we introduced review years and dates such as 1696 to simulate a manual entry error. To ensure that these timestamps are read correctly, this configuration was necessary. See [Spark issue SPARK-31404](https://issues.apache.org/jira/browse/SPARK-31404) about the calendar switch in the version 3.0.\n",
    "\n",
    "For a detailed explanation about these parameters, see [Spark Configuration](https://spark.apache.org/docs/latest/configuration.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      ":: loading settings :: url = jar:file:/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pyspark/jars/ivy-2.5.0.jar!/org/apache/ivy/core/settings/ivysettings.xml\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Ivy Default Cache set to: /home/ec2-user/.ivy2/cache\n",
      "The jars for the packages stored in: /home/ec2-user/.ivy2/jars\n",
      "com.amazon.deequ#deequ added as a dependency\n",
      ":: resolving dependencies :: org.apache.spark#spark-submit-parent-eb3a177b-ccc4-4677-a9b0-104653b54fc7;1.0\n",
      "\tconfs: [default]\n",
      "\tfound com.amazon.deequ#deequ;2.0.3-spark-3.3 in central\n",
      "\tfound org.scala-lang#scala-reflect;2.12.10 in central\n",
      "\tfound org.scalanlp#breeze_2.12;0.13.2 in central\n",
      "\tfound org.scalanlp#breeze-macros_2.12;0.13.2 in central\n",
      "\tfound com.github.fommil.netlib#core;1.1.2 in central\n",
      "\tfound net.sf.opencsv#opencsv;2.3 in central\n",
      "\tfound com.github.rwl#jtransforms;2.4.0 in central\n",
      "\tfound junit#junit;4.8.2 in central\n",
      "\tfound org.apache.commons#commons-math3;3.2 in central\n",
      "\tfound org.spire-math#spire_2.12;0.13.0 in central\n",
      "\tfound org.spire-math#spire-macros_2.12;0.13.0 in central\n",
      "\tfound org.typelevel#machinist_2.12;0.6.1 in central\n",
      "\tfound com.chuusai#shapeless_2.12;2.3.2 in central\n",
      "\tfound org.typelevel#macro-compat_2.12;1.1.1 in central\n",
      "\tfound org.slf4j#slf4j-api;1.7.5 in central\n",
      ":: resolution report :: resolve 493ms :: artifacts dl 18ms\n",
      "\t:: modules in use:\n",
      "\tcom.amazon.deequ#deequ;2.0.3-spark-3.3 from central in [default]\n",
      "\tcom.chuusai#shapeless_2.12;2.3.2 from central in [default]\n",
      "\tcom.github.fommil.netlib#core;1.1.2 from central in [default]\n",
      "\tcom.github.rwl#jtransforms;2.4.0 from central in [default]\n",
      "\tjunit#junit;4.8.2 from central in [default]\n",
      "\tnet.sf.opencsv#opencsv;2.3 from central in [default]\n",
      "\torg.apache.commons#commons-math3;3.2 from central in [default]\n",
      "\torg.scala-lang#scala-reflect;2.12.10 from central in [default]\n",
      "\torg.scalanlp#breeze-macros_2.12;0.13.2 from central in [default]\n",
      "\torg.scalanlp#breeze_2.12;0.13.2 from central in [default]\n",
      "\torg.slf4j#slf4j-api;1.7.5 from central in [default]\n",
      "\torg.spire-math#spire-macros_2.12;0.13.0 from central in [default]\n",
      "\torg.spire-math#spire_2.12;0.13.0 from central in [default]\n",
      "\torg.typelevel#machinist_2.12;0.6.1 from central in [default]\n",
      "\torg.typelevel#macro-compat_2.12;1.1.1 from central in [default]\n",
      "\t:: evicted modules:\n",
      "\torg.scala-lang#scala-reflect;2.12.1 by [org.scala-lang#scala-reflect;2.12.10] in [default]\n",
      "\torg.scala-lang#scala-reflect;2.12.0 by [org.scala-lang#scala-reflect;2.12.10] in [default]\n",
      "\t---------------------------------------------------------------------\n",
      "\t|                  |            modules            ||   artifacts   |\n",
      "\t|       conf       | number| search|dwnlded|evicted|| number|dwnlded|\n",
      "\t---------------------------------------------------------------------\n",
      "\t|      default     |   17  |   0   |   0   |   2   ||   15  |   0   |\n",
      "\t---------------------------------------------------------------------\n",
      ":: retrieving :: org.apache.spark#spark-submit-parent-eb3a177b-ccc4-4677-a9b0-104653b54fc7\n",
      "\tconfs: [default]\n",
      "\t0 artifacts copied, 15 already retrieved (0kB/13ms)\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "24/06/14 23:15:30 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SparkSession, Row, DataFrame\n",
    "import json\n",
    "import pandas as pd\n",
    "\n",
    "classpath = \":\".join(sagemaker_pyspark.classpath_jars())\n",
    "\n",
    "spark = (SparkSession\n",
    "    .builder\n",
    "    .config(\"spark.driver.extraClassPath\", classpath)\n",
    "    .config(\"spark.jars.packages\", pydeequ.deequ_maven_coord)\n",
    "    .config(\"spark.jars.excludes\", pydeequ.f2j_maven_coord)\n",
    "    .config(\"spark.driver.memory\", \"15g\")\n",
    "    .config(\"spark.sql.parquet.int96RebaseModeInRead\", \"CORRECTED\")\n",
    "    .getOrCreate())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read the dataset "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "24/06/14 23:15:33 WARN MetricsConfig: Cannot locate configuration: tried hadoop-metrics2-s3a-file-system.properties,hadoop-metrics2.properties\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df = spark.read.parquet(\"s3a://aws-bigdata-blog/generated_synthetic_reviews/data/product_category=Electronics/\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After you load the DataFrame, you can run `df.printSchema()` to view the schema of the dataset:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- marketplace: string (nullable = true)\n",
      " |-- customer_id: string (nullable = true)\n",
      " |-- review_id: string (nullable = true)\n",
      " |-- product_id: string (nullable = true)\n",
      " |-- product_title: string (nullable = true)\n",
      " |-- star_rating: long (nullable = true)\n",
      " |-- helpful_votes: long (nullable = true)\n",
      " |-- total_votes: long (nullable = true)\n",
      " |-- insight: string (nullable = true)\n",
      " |-- review_headline: string (nullable = true)\n",
      " |-- review_body: string (nullable = true)\n",
      " |-- review_date: timestamp (nullable = true)\n",
      " |-- review_year: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data Analysis \n",
    "\n",
    "Before we define checks on the data, we want to calculate some statistics for the dataset. As with Deequ, PyDeequ supports a rich set of metrics. For more information, see [Test data quality at scale with Deequ](https://aws.amazon.com/blogs/big-data/test-data-quality-at-scale-with-deequ/) or the [GitHub repo](https://github.com/awslabs/deequ/tree/master/src/main/scala/com/amazon/deequ/analyzers). In the following example, we use the [AnalysisRunner](https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/analyzers/runners/AnalysisRunner.scala) to capture the metrics we’re interested in:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pyspark/sql/dataframe.py:127: UserWarning: DataFrame constructor is internal. Do not directly use it.\n",
      "  warnings.warn(\"DataFrame constructor is internal. Do not directly use it.\")\n"
     ]
    }
   ],
   "source": [
    "from pydeequ.analyzers import *\n",
    "\n",
    "analysisResult = AnalysisRunner(spark) \\\n",
    "                    .onData(df) \\\n",
    "                    .addAnalyzer(Size()) \\\n",
    "                    .addAnalyzer(Completeness(\"review_id\")) \\\n",
    "                    .addAnalyzer(Distinctness(\"review_id\")) \\\n",
    "                    .addAnalyzer(Mean(\"star_rating\")) \\\n",
    "                    .addAnalyzer(Compliance(\"top star_rating\", \"star_rating >= 4.0\")) \\\n",
    "                    .addAnalyzer(Correlation(\"total_votes\", \"star_rating\")) \\\n",
    "                    .addAnalyzer(Correlation(\"total_votes\", \"helpful_votes\")) \\\n",
    "                    .run()\n",
    "                    \n",
    "analysisResult_df = AnalyzerContext.successMetricsAsDataFrame(spark, analysisResult)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------------------+------------+--------------------+\n",
      "|     entity|            instance|        name|               value|\n",
      "+-----------+--------------------+------------+--------------------+\n",
      "|     Column|           review_id|Completeness|                 1.0|\n",
      "|Mutlicolumn|total_votes,star_...| Correlation|-7.38808965018615...|\n",
      "|     Column|           review_id|Distinctness|  0.9926568563241371|\n",
      "|    Dataset|                   *|        Size|           3010972.0|\n",
      "|     Column|         star_rating|        Mean|  3.9999973430506826|\n",
      "|     Column|     top star_rating|  Compliance|  0.7499993357626706|\n",
      "|Mutlicolumn|total_votes,helpf...| Correlation|  0.9817922803462663|\n",
      "+-----------+--------------------+------------+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pd.options.display.float_format = '{:,.7g}'.format\n",
    "analysisResult_df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From this, we learn the following:\n",
    "\n",
    "- `review_id` has no missing values and approximately 99.27% of the values are distinct\n",
    "- 74.99% of reviews have a `star_rating` of 4 or higher\n",
    "- `total_votes` and `star_rating` are not correlated\n",
    "- `helpful_votes` and `total_votes` are strongly correlated\n",
    "- The average `star_rating` is 3.99\n",
    "- The dataset contains 3,010,972 reviews\n",
    "\n",
    "Sometimes, you may want to run multiple metrics on a single column. For example, you want to check that all reviews were written either after 1996 or before 2017. In this case, it’s helpful to provide a name for each metric in order to distinguish the results in the output:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>entity</th>\n",
       "      <th>instance</th>\n",
       "      <th>name</th>\n",
       "      <th>value</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Column</td>\n",
       "      <td>after-1996 review_year</td>\n",
       "      <td>Compliance</td>\n",
       "      <td>0.9999993</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Column</td>\n",
       "      <td>before-2017 review_year</td>\n",
       "      <td>Compliance</td>\n",
       "      <td>0.999999</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   entity                 instance        name     value\n",
       "0  Column   after-1996 review_year  Compliance 0.9999993\n",
       "1  Column  before-2017 review_year  Compliance  0.999999"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "analysisResult = AnalysisRunner(spark) \\\n",
    "                    .onData(df) \\\n",
    "                    .addAnalyzer(Compliance(\"after-1996 review_year\", \n",
    "\"review_year >= 1996\")) \\\n",
    "                    .addAnalyzer(Compliance(\"before-2017 review_year\", \n",
    "\"review_year <= 2017\")) \\\n",
    "                    .run()\n",
    "analysisResult_pd_df = AnalyzerContext.successMetricsAsDataFrame(spark,\n",
    "analysisResult, pandas=True)\n",
    "analysisResult_pd_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alternatively, you can combine the conditions by using logical operators `and` and `or`. In the following example, we check that the years for the reviews are between 1996 and 2017 and the values for the `insight` column are either 'Y' or 'N':"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pyspark/sql/dataframe.py:127: UserWarning: DataFrame constructor is internal. Do not directly use it.\n",
      "  warnings.warn(\"DataFrame constructor is internal. Do not directly use it.\")\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>entity</th>\n",
       "      <th>instance</th>\n",
       "      <th>name</th>\n",
       "      <th>value</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Column</td>\n",
       "      <td>range1996to2017 review_year</td>\n",
       "      <td>Compliance</td>\n",
       "      <td>0.9999983</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>Column</td>\n",
       "      <td>values insight</td>\n",
       "      <td>Compliance</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   entity                     instance        name     value\n",
       "0  Column  range1996to2017 review_year  Compliance 0.9999983\n",
       "1  Column               values insight  Compliance         1"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "analysisResult = AnalysisRunner(spark) \\\n",
    "                    .onData(df) \\\n",
    "                    .addAnalyzer(Compliance(\"range1996to2017 review_year\",\n",
    "\"review_year >= 1996 and review_year <= 2017\")) \\\n",
    "                    .addAnalyzer(Compliance(\"values insight\", \n",
    "\"insight == 'Y' or insight == 'N'\")) \\\n",
    "                    .run()\n",
    "analysisResult_pd_df = AnalyzerContext.successMetricsAsDataFrame(spark, analysisResult, pandas=True)\n",
    "analysisResult_pd_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In some cases, you might prefer a different format for the output. PyDeequ allows you to output the results in a JSON format:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'entity': 'Column',\n",
       "  'instance': 'range1996to2017 review_year',\n",
       "  'name': 'Compliance',\n",
       "  'value': 0.9999983394066766},\n",
       " {'entity': 'Column',\n",
       "  'instance': 'values insight',\n",
       "  'name': 'Compliance',\n",
       "  'value': 1.0}]"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "analysisResult_json = AnalyzerContext.successMetricsAsJson(spark, analysisResult)\n",
    "analysisResult_json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define and Run Tests for Data\n",
    "\n",
    "After analyzing and understanding the data, we want to verify that the properties we have derived also hold for new versions of the dataset. By defining assertions on the data distribution as part of a data pipeline, we can make sure every processed dataset is of high quality, and that any application consuming the data can rely on it.\n",
    "\n",
    "For writing tests on data, we start with the `VerificationSuite` and add [checks](https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/checks/Check.scala) on attributes of the data. In this example, we test for the following properties of our data:\n",
    "\n",
    "- At least 3 million rows in total\n",
    "- `review_id` is never null\n",
    "- `review_id` is unique\n",
    "- `star_rating` has a minimum of 1.0 and maximum of 5.0\n",
    "- `marketplace` only contains `US`, `UK`, `DE`, `JP`, or `FR`\n",
    "- `year` does not contain negative values\n",
    "- `year` is between 1996 and 2017\n",
    "\n",
    "The following code reflects the previous statements. For information about all available checks, see the [GitHub repo](https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/checks/Check.scala). You can run this directly in the Spark shell as previously explained:\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Python Callback server started!\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pyspark/sql/dataframe.py:127: UserWarning: DataFrame constructor is internal. Do not directly use it.\n",
      "  warnings.warn(\"DataFrame constructor is internal. Do not directly use it.\")\n"
     ]
    }
   ],
   "source": [
    "from pydeequ.checks import *\n",
    "from pydeequ.verification import *\n",
    "\n",
    "check = Check(spark, CheckLevel.Warning, \"Synthetic Product Reviews\")\n",
    "\n",
    "checkResult = VerificationSuite(spark) \\\n",
    "    .onData(df) \\\n",
    "    .addCheck(\n",
    "        check.hasSize(lambda x: x >= 3000000) \\\n",
    "        .hasMin(\"star_rating\", lambda x: x == 1.0) \\\n",
    "        .hasMax(\"star_rating\", lambda x: x == 5.0)  \\\n",
    "        .isComplete(\"review_id\")  \\\n",
    "        .isUnique(\"review_id\")  \\\n",
    "        .isComplete(\"marketplace\")  \\\n",
    "        .isContainedIn(\"marketplace\", [\"US\", \"UK\", \"DE\", \"JP\", \"FR\"]) \\\n",
    "        .isNonNegative(\"year\") \\\n",
    "        .hasMin(\"review_year\", lambda x: x == '1996') \\\n",
    "        .hasMax(\"review_year\", lambda x: x == '2017')) \\\n",
    "    .run()\n",
    "    \n",
    "checkResult_df = VerificationResult.checkResultsAsDataFrame(spark,\n",
    "checkResult, pandas=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we change the display settings of the DataFrame to ensure that the entire constraint message is visible."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style type=\"text/css\">\n",
       "#T_11617_row0_col0, #T_11617_row0_col1, #T_11617_row0_col2, #T_11617_row0_col3, #T_11617_row0_col4, #T_11617_row0_col5, #T_11617_row1_col0, #T_11617_row1_col1, #T_11617_row1_col2, #T_11617_row1_col3, #T_11617_row1_col4, #T_11617_row1_col5, #T_11617_row2_col0, #T_11617_row2_col1, #T_11617_row2_col2, #T_11617_row2_col3, #T_11617_row2_col4, #T_11617_row2_col5, #T_11617_row3_col0, #T_11617_row3_col1, #T_11617_row3_col2, #T_11617_row3_col3, #T_11617_row3_col4, #T_11617_row3_col5, #T_11617_row4_col0, #T_11617_row4_col1, #T_11617_row4_col2, #T_11617_row4_col3, #T_11617_row4_col4, #T_11617_row4_col5, #T_11617_row5_col0, #T_11617_row5_col1, #T_11617_row5_col2, #T_11617_row5_col3, #T_11617_row5_col4, #T_11617_row5_col5, #T_11617_row6_col0, #T_11617_row6_col1, #T_11617_row6_col2, #T_11617_row6_col3, #T_11617_row6_col4, #T_11617_row6_col5, #T_11617_row7_col0, #T_11617_row7_col1, #T_11617_row7_col2, #T_11617_row7_col3, #T_11617_row7_col4, #T_11617_row7_col5, #T_11617_row8_col0, #T_11617_row8_col1, #T_11617_row8_col2, #T_11617_row8_col3, #T_11617_row8_col4, #T_11617_row8_col5, #T_11617_row9_col0, #T_11617_row9_col1, #T_11617_row9_col2, #T_11617_row9_col3, #T_11617_row9_col4, #T_11617_row9_col5 {\n",
       "  overflow-wrap: break-word;\n",
       "  inline-size: 10px;\n",
       "}\n",
       "</style>\n",
       "<table id=\"T_11617\">\n",
       "  <thead>\n",
       "    <tr>\n",
       "      <th class=\"blank level0\" >&nbsp;</th>\n",
       "      <th id=\"T_11617_level0_col0\" class=\"col_heading level0 col0\" >check</th>\n",
       "      <th id=\"T_11617_level0_col1\" class=\"col_heading level0 col1\" >check_level</th>\n",
       "      <th id=\"T_11617_level0_col2\" class=\"col_heading level0 col2\" >check_status</th>\n",
       "      <th id=\"T_11617_level0_col3\" class=\"col_heading level0 col3\" >constraint</th>\n",
       "      <th id=\"T_11617_level0_col4\" class=\"col_heading level0 col4\" >constraint_status</th>\n",
       "      <th id=\"T_11617_level0_col5\" class=\"col_heading level0 col5\" >constraint_message</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row0\" class=\"row_heading level0 row0\" >0</th>\n",
       "      <td id=\"T_11617_row0_col0\" class=\"data row0 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row0_col1\" class=\"data row0 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row0_col2\" class=\"data row0 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row0_col3\" class=\"data row0 col3\" >SizeConstraint(Size(None))</td>\n",
       "      <td id=\"T_11617_row0_col4\" class=\"data row0 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row0_col5\" class=\"data row0 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row1\" class=\"row_heading level0 row1\" >1</th>\n",
       "      <td id=\"T_11617_row1_col0\" class=\"data row1 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row1_col1\" class=\"data row1 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row1_col2\" class=\"data row1 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row1_col3\" class=\"data row1 col3\" >MinimumConstraint(Minimum(star_rating,None))</td>\n",
       "      <td id=\"T_11617_row1_col4\" class=\"data row1 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row1_col5\" class=\"data row1 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row2\" class=\"row_heading level0 row2\" >2</th>\n",
       "      <td id=\"T_11617_row2_col0\" class=\"data row2 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row2_col1\" class=\"data row2 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row2_col2\" class=\"data row2 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row2_col3\" class=\"data row2 col3\" >MaximumConstraint(Maximum(star_rating,None))</td>\n",
       "      <td id=\"T_11617_row2_col4\" class=\"data row2 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row2_col5\" class=\"data row2 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row3\" class=\"row_heading level0 row3\" >3</th>\n",
       "      <td id=\"T_11617_row3_col0\" class=\"data row3 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row3_col1\" class=\"data row3 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row3_col2\" class=\"data row3 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row3_col3\" class=\"data row3 col3\" >CompletenessConstraint(Completeness(review_id,None))</td>\n",
       "      <td id=\"T_11617_row3_col4\" class=\"data row3 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row3_col5\" class=\"data row3 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row4\" class=\"row_heading level0 row4\" >4</th>\n",
       "      <td id=\"T_11617_row4_col0\" class=\"data row4 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row4_col1\" class=\"data row4 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row4_col2\" class=\"data row4 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row4_col3\" class=\"data row4 col3\" >UniquenessConstraint(Uniqueness(List(review_id),None))</td>\n",
       "      <td id=\"T_11617_row4_col4\" class=\"data row4 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row4_col5\" class=\"data row4 col5\" >Value: 0.9853137126482744 does not meet the constraint requirement!</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row5\" class=\"row_heading level0 row5\" >5</th>\n",
       "      <td id=\"T_11617_row5_col0\" class=\"data row5 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row5_col1\" class=\"data row5 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row5_col2\" class=\"data row5 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row5_col3\" class=\"data row5 col3\" >CompletenessConstraint(Completeness(marketplace,None))</td>\n",
       "      <td id=\"T_11617_row5_col4\" class=\"data row5 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row5_col5\" class=\"data row5 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row6\" class=\"row_heading level0 row6\" >6</th>\n",
       "      <td id=\"T_11617_row6_col0\" class=\"data row6 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row6_col1\" class=\"data row6 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row6_col2\" class=\"data row6 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row6_col3\" class=\"data row6 col3\" >ComplianceConstraint(Compliance(marketplace contained in US,UK,DE,JP,FR,`marketplace` IS NULL OR `marketplace` IN ('US','UK','DE','JP','FR'),None))</td>\n",
       "      <td id=\"T_11617_row6_col4\" class=\"data row6 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row6_col5\" class=\"data row6 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row7\" class=\"row_heading level0 row7\" >7</th>\n",
       "      <td id=\"T_11617_row7_col0\" class=\"data row7 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row7_col1\" class=\"data row7 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row7_col2\" class=\"data row7 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row7_col3\" class=\"data row7 col3\" >ComplianceConstraint(Compliance(year is non-negative,COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0,None))</td>\n",
       "      <td id=\"T_11617_row7_col4\" class=\"data row7 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row7_col5\" class=\"data row7 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row8\" class=\"row_heading level0 row8\" >8</th>\n",
       "      <td id=\"T_11617_row8_col0\" class=\"data row8 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row8_col1\" class=\"data row8 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row8_col2\" class=\"data row8 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row8_col3\" class=\"data row8 col3\" >MinimumConstraint(Minimum(review_year,None))</td>\n",
       "      <td id=\"T_11617_row8_col4\" class=\"data row8 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row8_col5\" class=\"data row8 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th id=\"T_11617_level0_row9\" class=\"row_heading level0 row9\" >9</th>\n",
       "      <td id=\"T_11617_row9_col0\" class=\"data row9 col0\" >Synthetic Product Reviews</td>\n",
       "      <td id=\"T_11617_row9_col1\" class=\"data row9 col1\" >Warning</td>\n",
       "      <td id=\"T_11617_row9_col2\" class=\"data row9 col2\" >Warning</td>\n",
       "      <td id=\"T_11617_row9_col3\" class=\"data row9 col3\" >MaximumConstraint(Maximum(review_year,None))</td>\n",
       "      <td id=\"T_11617_row9_col4\" class=\"data row9 col4\" >Failure</td>\n",
       "      <td id=\"T_11617_row9_col5\" class=\"data row9 col5\" >org.apache.spark.sql.AnalysisException: Column 'year' does not exist. Did you mean one of the following? [insight, review_year, review_id, marketplace, review_date, star_rating, customer_id, product_id, review_body, total_votes, helpful_votes, product_title, review_headline]; line 1 pos 14;\n",
       "'Aggregate [sum(cast(isnotnull(review_id#2) as int)) AS sum(CAST((review_id IS NOT NULL) AS INT))#629L, count(1) AS count(1)#630L, cast(min(review_year#12L) as double) AS CAST(min(review_year) AS DOUBLE)#631, cast(max(review_year#12L) as double) AS CAST(max(review_year) AS DOUBLE)#632, count(1) AS count(1)#633L, cast(min(star_rating#5L) as double) AS CAST(min(star_rating) AS DOUBLE)#634, cast(max(star_rating#5L) as double) AS CAST(max(star_rating) AS DOUBLE)#635, sum(cast((isnull(marketplace#0) OR marketplace#0 IN (US,UK,DE,JP,FR)) as int)) AS sum(CAST(((marketplace IS NULL) OR (marketplace IN (US, UK, DE, JP, FR))) AS INT))#636L, count(1) AS count(1)#637L, sum(cast(isnotnull(marketplace#0) as int)) AS sum(CAST((marketplace IS NOT NULL) AS INT))#638L, count(1) AS count(1)#639L, sum(cast(('COALESCE(cast('year as decimal(20,10)), 0.0) >= 0) as int)) AS sum(CAST((COALESCE(CAST(year AS DECIMAL(20,10)), 0.0) >= 0) AS INT))#640, count(1) AS count(1)#641L]\n",
       "+- Relation [marketplace#0,customer_id#1,review_id#2,product_id#3,product_title#4,star_rating#5L,helpful_votes#6L,total_votes#7L,insight#8,review_headline#9,review_body#10,review_date#11,review_year#12L] parquet\n",
       "</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n"
      ],
      "text/plain": [
       "<pandas.io.formats.style.Styler at 0x7f1925bf3550>"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "checkResult_df.style.set_properties(\n",
    "    **{\n",
    "        'overflow-wrap': 'break-word',\n",
    "        'inline-size': '10px',\n",
    "    }\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After calling `run()`, PyDeequ translates your test description into Deequ, which translates it into a series of Spark jobs that are run to compute metrics on the data. Afterwards, it invokes your assertion functions (for example, `lambda x: x == 1.0` for the minimum star rating check) on these metrics to see if the constraints hold on the data. \n",
    "\n",
    "Interestingly, the `review_id` column isn’t unique, which resulted in a failure of the check on uniqueness. We can also look at all the metrics that Deequ computed for this check by running the following:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/ec2-user/anaconda3/envs/python3/lib/python3.10/site-packages/pyspark/sql/dataframe.py:127: UserWarning: DataFrame constructor is internal. Do not directly use it.\n",
      "  warnings.warn(\"DataFrame constructor is internal. Do not directly use it.\")\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>entity</th>\n",
       "      <th>instance</th>\n",
       "      <th>name</th>\n",
       "      <th>value</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>Column</td>\n",
       "      <td>review_id</td>\n",
       "      <td>Uniqueness</td>\n",
       "      <td>0.9853137</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   entity   instance        name     value\n",
       "0  Column  review_id  Uniqueness 0.9853137"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "checkResults_df = VerificationResult.successMetricsAsDataFrame(spark, checkResult, pandas = True)\n",
    "checkResults_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Automated Constraint Suggestion \n",
    "\n",
    "If you own a large number of datasets or if your dataset has many columns, it may be challenging for you to manually define appropriate constraints. Deequ can automatically suggest useful constraints based on the data distribution. Deequ first runs a data profiling method and then applies a set of rules on the result. For more information about how to run a data profiling method, see the [GitHub repo](https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/data_profiling_example.md)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "24/06/14 23:15:56 WARN package: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 34:==================================================>     (10 + 1) / 11]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "  \"constraint_suggestions\": [\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('insight' has value range 'N', 'Y',`insight` IN ('N', 'Y'),None))\",\n",
      "      \"column_name\": \"insight\",\n",
      "      \"current_value\": \"Compliance: 1\",\n",
      "      \"description\": \"'insight' has value range 'N', 'Y'\",\n",
      "      \"suggesting_rule\": \"CategoricalRangeRule(com.amazon.deequ.suggestions.rules.CategoricalRangeRule$$$Lambda$4119/0x000000080197e840@74f276b0)\",\n",
      "      \"rule_description\": \"If we see a categorical range for a column, we suggest an IS IN (...) constraint\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"insight\\\", [\\\"N\\\", \\\"Y\\\"])\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(insight,None))\",\n",
      "      \"column_name\": \"insight\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'insight' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"insight\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(review_id,None))\",\n",
      "      \"column_name\": \"review_id\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'review_id' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"review_id\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"UniquenessConstraint(Uniqueness(List(review_id),None))\",\n",
      "      \"column_name\": \"review_id\",\n",
      "      \"current_value\": \"ApproxDistinctness: 1.0496308168923523\",\n",
      "      \"description\": \"'review_id' is unique\",\n",
      "      \"suggesting_rule\": \"UniqueIfApproximatelyUniqueRule()\",\n",
      "      \"rule_description\": \"If the ratio of approximate num distinct values in a column is close to the number of records (within the error of the HLL sketch), we suggest a UNIQUE constraint\",\n",
      "      \"code_for_constraint\": \".isUnique(\\\"review_id\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(customer_id,None))\",\n",
      "      \"column_name\": \"customer_id\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'customer_id' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"customer_id\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('customer_id' has no negative values,customer_id >= 0,None))\",\n",
      "      \"column_name\": \"customer_id\",\n",
      "      \"current_value\": \"Minimum: 100000.0\",\n",
      "      \"description\": \"'customer_id' has no negative values\",\n",
      "      \"suggesting_rule\": \"NonNegativeNumbersRule()\",\n",
      "      \"rule_description\": \"If we see only non-negative numbers in a column, we suggest a corresponding constraint\",\n",
      "      \"code_for_constraint\": \".isNonNegative(\\\"customer_id\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"AnalysisBasedConstraint(DataType(customer_id,None),<function1>,Some(<function1>),None)\",\n",
      "      \"column_name\": \"customer_id\",\n",
      "      \"current_value\": \"DataType: Integral\",\n",
      "      \"description\": \"'customer_id' has type Integral\",\n",
      "      \"suggesting_rule\": \"RetainTypeRule()\",\n",
      "      \"rule_description\": \"If we detect a non-string type, we suggest a type constraint\",\n",
      "      \"code_for_constraint\": \".hasDataType(\\\"customer_id\\\", ConstrainableDataTypes.Integral)\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(review_date,None))\",\n",
      "      \"column_name\": \"review_date\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'review_date' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"review_date\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('helpful_votes' has value range '15', '14', '13', '16', '17', '12', '18', '11', '19', '10', '9', '20', '8', '21', '22', '7', '6', '23', '5', '24', '4', '25', '26', '3',`helpful_votes` IN ('15', '14', '13', '16', '17', '12', '18', '11', '19', '10', '9', '20', '8', '21', '22', '7', '6', '23', '5', '24', '4', '25', '26', '3'),None))\",\n",
      "      \"column_name\": \"helpful_votes\",\n",
      "      \"current_value\": \"Compliance: 1\",\n",
      "      \"description\": \"'helpful_votes' has value range '15', '14', '13', '16', '17', '12', '18', '11', '19', '10', '9', '20', '8', '21', '22', '7', '6', '23', '5', '24', '4', '25', '26', '3'\",\n",
      "      \"suggesting_rule\": \"CategoricalRangeRule(com.amazon.deequ.suggestions.rules.CategoricalRangeRule$$$Lambda$4119/0x000000080197e840@74f276b0)\",\n",
      "      \"rule_description\": \"If we see a categorical range for a column, we suggest an IS IN (...) constraint\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"helpful_votes\\\", [\\\"15\\\", \\\"14\\\", \\\"13\\\", \\\"16\\\", \\\"17\\\", \\\"12\\\", \\\"18\\\", \\\"11\\\", \\\"19\\\", \\\"10\\\", \\\"9\\\", \\\"20\\\", \\\"8\\\", \\\"21\\\", \\\"22\\\", \\\"7\\\", \\\"6\\\", \\\"23\\\", \\\"5\\\", \\\"24\\\", \\\"4\\\", \\\"25\\\", \\\"26\\\", \\\"3\\\"])\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(helpful_votes,None))\",\n",
      "      \"column_name\": \"helpful_votes\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'helpful_votes' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"helpful_votes\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('helpful_votes' has value range '15', '14', '13', '16', '17', '12', '18', '11' for at least 92.0% of values,`helpful_votes` IN ('15', '14', '13', '16', '17', '12', '18', '11'),None))\",\n",
      "      \"column_name\": \"helpful_votes\",\n",
      "      \"current_value\": \"Compliance: 0.926417449248947\",\n",
      "      \"description\": \"'helpful_votes' has value range '15', '14', '13', '16', '17', '12', '18', '11' for at least 92.0% of values\",\n",
      "      \"suggesting_rule\": \"FractionalCategoricalRangeRule(0.9,com.amazon.deequ.suggestions.rules.FractionalCategoricalRangeRule$$$Lambda$4120/0x000000080197f040@65f67611)\",\n",
      "      \"rule_description\": \"If we see a categorical range for most values in a column, we suggest an IS IN (...) constraint that should hold for most values\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"helpful_votes\\\", [\\\"15\\\", \\\"14\\\", \\\"13\\\", \\\"16\\\", \\\"17\\\", \\\"12\\\", \\\"18\\\", \\\"11\\\"], lambda x: x >= 0.92, \\\"It should be above 0.92!\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('helpful_votes' has no negative values,helpful_votes >= 0,None))\",\n",
      "      \"column_name\": \"helpful_votes\",\n",
      "      \"current_value\": \"Minimum: 3.0\",\n",
      "      \"description\": \"'helpful_votes' has no negative values\",\n",
      "      \"suggesting_rule\": \"NonNegativeNumbersRule()\",\n",
      "      \"rule_description\": \"If we see only non-negative numbers in a column, we suggest a corresponding constraint\",\n",
      "      \"code_for_constraint\": \".isNonNegative(\\\"helpful_votes\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('star_rating' has value range '5', '4', '3', '2', '1',`star_rating` IN ('5', '4', '3', '2', '1'),None))\",\n",
      "      \"column_name\": \"star_rating\",\n",
      "      \"current_value\": \"Compliance: 1\",\n",
      "      \"description\": \"'star_rating' has value range '5', '4', '3', '2', '1'\",\n",
      "      \"suggesting_rule\": \"CategoricalRangeRule(com.amazon.deequ.suggestions.rules.CategoricalRangeRule$$$Lambda$4119/0x000000080197e840@74f276b0)\",\n",
      "      \"rule_description\": \"If we see a categorical range for a column, we suggest an IS IN (...) constraint\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"star_rating\\\", [\\\"5\\\", \\\"4\\\", \\\"3\\\", \\\"2\\\", \\\"1\\\"])\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(star_rating,None))\",\n",
      "      \"column_name\": \"star_rating\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'star_rating' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"star_rating\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('star_rating' has value range '5', '4', '3', '2' for at least 94.0% of values,`star_rating` IN ('5', '4', '3', '2'),None))\",\n",
      "      \"column_name\": \"star_rating\",\n",
      "      \"current_value\": \"Compliance: 0.9499998671525341\",\n",
      "      \"description\": \"'star_rating' has value range '5', '4', '3', '2' for at least 94.0% of values\",\n",
      "      \"suggesting_rule\": \"FractionalCategoricalRangeRule(0.9,com.amazon.deequ.suggestions.rules.FractionalCategoricalRangeRule$$$Lambda$4120/0x000000080197f040@65f67611)\",\n",
      "      \"rule_description\": \"If we see a categorical range for most values in a column, we suggest an IS IN (...) constraint that should hold for most values\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"star_rating\\\", [\\\"5\\\", \\\"4\\\", \\\"3\\\", \\\"2\\\"], lambda x: x >= 0.94, \\\"It should be above 0.94!\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('star_rating' has no negative values,star_rating >= 0,None))\",\n",
      "      \"column_name\": \"star_rating\",\n",
      "      \"current_value\": \"Minimum: 1.0\",\n",
      "      \"description\": \"'star_rating' has no negative values\",\n",
      "      \"suggesting_rule\": \"NonNegativeNumbersRule()\",\n",
      "      \"rule_description\": \"If we see only non-negative numbers in a column, we suggest a corresponding constraint\",\n",
      "      \"code_for_constraint\": \".isNonNegative(\\\"star_rating\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(product_title,None))\",\n",
      "      \"column_name\": \"product_title\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'product_title' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"product_title\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(review_headline,None))\",\n",
      "      \"column_name\": \"review_headline\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'review_headline' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"review_headline\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('review_year' has value range '2016', '2015', '2014', '2013', '2012', '2011', '2010', '2009', '2008', '2007', '2006', '2005', '2004', '2003', '2002', '2001', '2000', '1999', '1998', '1997', '1996', '1696', '2101', '2202',`review_year` IN ('2016', '2015', '2014', '2013', '2012', '2011', '2010', '2009', '2008', '2007', '2006', '2005', '2004', '2003', '2002', '2001', '2000', '1999', '1998', '1997', '1996', '1696', '2101', '2202'),None))\",\n",
      "      \"column_name\": \"review_year\",\n",
      "      \"current_value\": \"Compliance: 1\",\n",
      "      \"description\": \"'review_year' has value range '2016', '2015', '2014', '2013', '2012', '2011', '2010', '2009', '2008', '2007', '2006', '2005', '2004', '2003', '2002', '2001', '2000', '1999', '1998', '1997', '1996', '1696', '2101', '2202'\",\n",
      "      \"suggesting_rule\": \"CategoricalRangeRule(com.amazon.deequ.suggestions.rules.CategoricalRangeRule$$$Lambda$4119/0x000000080197e840@74f276b0)\",\n",
      "      \"rule_description\": \"If we see a categorical range for a column, we suggest an IS IN (...) constraint\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"review_year\\\", [\\\"2016\\\", \\\"2015\\\", \\\"2014\\\", \\\"2013\\\", \\\"2012\\\", \\\"2011\\\", \\\"2010\\\", \\\"2009\\\", \\\"2008\\\", \\\"2007\\\", \\\"2006\\\", \\\"2005\\\", \\\"2004\\\", \\\"2003\\\", \\\"2002\\\", \\\"2001\\\", \\\"2000\\\", \\\"1999\\\", \\\"1998\\\", \\\"1997\\\", \\\"1996\\\", \\\"1696\\\", \\\"2101\\\", \\\"2202\\\"])\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(review_year,None))\",\n",
      "      \"column_name\": \"review_year\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'review_year' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"review_year\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('review_year' has value range '2016', '2015', '2014', '2013', '2012', '2011', '2010', '2009', '2008', '2007', '2006', '2005' for at least 91.0% of values,`review_year` IN ('2016', '2015', '2014', '2013', '2012', '2011', '2010', '2009', '2008', '2007', '2006', '2005'),None))\",\n",
      "      \"column_name\": \"review_year\",\n",
      "      \"current_value\": \"Compliance: 0.9157531189263799\",\n",
      "      \"description\": \"'review_year' has value range '2016', '2015', '2014', '2013', '2012', '2011', '2010', '2009', '2008', '2007', '2006', '2005' for at least 91.0% of values\",\n",
      "      \"suggesting_rule\": \"FractionalCategoricalRangeRule(0.9,com.amazon.deequ.suggestions.rules.FractionalCategoricalRangeRule$$$Lambda$4120/0x000000080197f040@65f67611)\",\n",
      "      \"rule_description\": \"If we see a categorical range for most values in a column, we suggest an IS IN (...) constraint that should hold for most values\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"review_year\\\", [\\\"2016\\\", \\\"2015\\\", \\\"2014\\\", \\\"2013\\\", \\\"2012\\\", \\\"2011\\\", \\\"2010\\\", \\\"2009\\\", \\\"2008\\\", \\\"2007\\\", \\\"2006\\\", \\\"2005\\\"], lambda x: x >= 0.91, \\\"It should be above 0.91!\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('review_year' has no negative values,review_year >= 0,None))\",\n",
      "      \"column_name\": \"review_year\",\n",
      "      \"current_value\": \"Minimum: 1696.0\",\n",
      "      \"description\": \"'review_year' has no negative values\",\n",
      "      \"suggesting_rule\": \"NonNegativeNumbersRule()\",\n",
      "      \"rule_description\": \"If we see only non-negative numbers in a column, we suggest a corresponding constraint\",\n",
      "      \"code_for_constraint\": \".isNonNegative(\\\"review_year\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(product_id,None))\",\n",
      "      \"column_name\": \"product_id\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'product_id' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"product_id\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('product_id' has no negative values,product_id >= 0,None))\",\n",
      "      \"column_name\": \"product_id\",\n",
      "      \"current_value\": \"Minimum: 10000.0\",\n",
      "      \"description\": \"'product_id' has no negative values\",\n",
      "      \"suggesting_rule\": \"NonNegativeNumbersRule()\",\n",
      "      \"rule_description\": \"If we see only non-negative numbers in a column, we suggest a corresponding constraint\",\n",
      "      \"code_for_constraint\": \".isNonNegative(\\\"product_id\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"AnalysisBasedConstraint(DataType(product_id,None),<function1>,Some(<function1>),None)\",\n",
      "      \"column_name\": \"product_id\",\n",
      "      \"current_value\": \"DataType: Integral\",\n",
      "      \"description\": \"'product_id' has type Integral\",\n",
      "      \"suggesting_rule\": \"RetainTypeRule()\",\n",
      "      \"rule_description\": \"If we detect a non-string type, we suggest a type constraint\",\n",
      "      \"code_for_constraint\": \".hasDataType(\\\"product_id\\\", ConstrainableDataTypes.Integral)\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('total_votes' has value range '19', '20', '21', '18', '17', '22', '23', '16', '15', '24', '25', '14', '26', '13', '12', '27', '28', '11', '10', '29', '30', '9', '8', '31', '32', '7', '33', '6', '5', '34', '4', '35', '3', '36', '2', '37', '38', '39', '1', '0',`total_votes` IN ('19', '20', '21', '18', '17', '22', '23', '16', '15', '24', '25', '14', '26', '13', '12', '27', '28', '11', '10', '29', '30', '9', '8', '31', '32', '7', '33', '6', '5', '34', '4', '35', '3', '36', '2', '37', '38', '39', '1', '0'),None))\",\n",
      "      \"column_name\": \"total_votes\",\n",
      "      \"current_value\": \"Compliance: 1\",\n",
      "      \"description\": \"'total_votes' has value range '19', '20', '21', '18', '17', '22', '23', '16', '15', '24', '25', '14', '26', '13', '12', '27', '28', '11', '10', '29', '30', '9', '8', '31', '32', '7', '33', '6', '5', '34', '4', '35', '3', '36', '2', '37', '38', '39', '1', '0'\",\n",
      "      \"suggesting_rule\": \"CategoricalRangeRule(com.amazon.deequ.suggestions.rules.CategoricalRangeRule$$$Lambda$4119/0x000000080197e840@74f276b0)\",\n",
      "      \"rule_description\": \"If we see a categorical range for a column, we suggest an IS IN (...) constraint\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"total_votes\\\", [\\\"19\\\", \\\"20\\\", \\\"21\\\", \\\"18\\\", \\\"17\\\", \\\"22\\\", \\\"23\\\", \\\"16\\\", \\\"15\\\", \\\"24\\\", \\\"25\\\", \\\"14\\\", \\\"26\\\", \\\"13\\\", \\\"12\\\", \\\"27\\\", \\\"28\\\", \\\"11\\\", \\\"10\\\", \\\"29\\\", \\\"30\\\", \\\"9\\\", \\\"8\\\", \\\"31\\\", \\\"32\\\", \\\"7\\\", \\\"33\\\", \\\"6\\\", \\\"5\\\", \\\"34\\\", \\\"4\\\", \\\"35\\\", \\\"3\\\", \\\"36\\\", \\\"2\\\", \\\"37\\\", \\\"38\\\", \\\"39\\\", \\\"1\\\", \\\"0\\\"])\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(total_votes,None))\",\n",
      "      \"column_name\": \"total_votes\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'total_votes' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"total_votes\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('total_votes' has value range '19', '20', '21', '18', '17', '22', '23', '16', '15', '24', '25', '14', '26' for at least 90.0% of values,`total_votes` IN ('19', '20', '21', '18', '17', '22', '23', '16', '15', '24', '25', '14', '26'),None))\",\n",
      "      \"column_name\": \"total_votes\",\n",
      "      \"current_value\": \"Compliance: 0.904062874048646\",\n",
      "      \"description\": \"'total_votes' has value range '19', '20', '21', '18', '17', '22', '23', '16', '15', '24', '25', '14', '26' for at least 90.0% of values\",\n",
      "      \"suggesting_rule\": \"FractionalCategoricalRangeRule(0.9,com.amazon.deequ.suggestions.rules.FractionalCategoricalRangeRule$$$Lambda$4120/0x000000080197f040@65f67611)\",\n",
      "      \"rule_description\": \"If we see a categorical range for most values in a column, we suggest an IS IN (...) constraint that should hold for most values\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"total_votes\\\", [\\\"19\\\", \\\"20\\\", \\\"21\\\", \\\"18\\\", \\\"17\\\", \\\"22\\\", \\\"23\\\", \\\"16\\\", \\\"15\\\", \\\"24\\\", \\\"25\\\", \\\"14\\\", \\\"26\\\"], lambda x: x >= 0.9, \\\"It should be above 0.9!\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('total_votes' has no negative values,total_votes >= 0,None))\",\n",
      "      \"column_name\": \"total_votes\",\n",
      "      \"current_value\": \"Minimum: 0.0\",\n",
      "      \"description\": \"'total_votes' has no negative values\",\n",
      "      \"suggesting_rule\": \"NonNegativeNumbersRule()\",\n",
      "      \"rule_description\": \"If we see only non-negative numbers in a column, we suggest a corresponding constraint\",\n",
      "      \"code_for_constraint\": \".isNonNegative(\\\"total_votes\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(review_body,None))\",\n",
      "      \"column_name\": \"review_body\",\n",
      "      \"current_value\": \"Completeness: 1.0\",\n",
      "      \"description\": \"'review_body' is not null\",\n",
      "      \"suggesting_rule\": \"CompleteIfCompleteRule()\",\n",
      "      \"rule_description\": \"If a column is complete in the sample, we suggest a NOT NULL constraint\",\n",
      "      \"code_for_constraint\": \".isComplete(\\\"review_body\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"UniquenessConstraint(Uniqueness(List(review_body),None))\",\n",
      "      \"column_name\": \"review_body\",\n",
      "      \"current_value\": \"ApproxDistinctness: 1.0332703193520232\",\n",
      "      \"description\": \"'review_body' is unique\",\n",
      "      \"suggesting_rule\": \"UniqueIfApproximatelyUniqueRule()\",\n",
      "      \"rule_description\": \"If the ratio of approximate num distinct values in a column is close to the number of records (within the error of the HLL sketch), we suggest a UNIQUE constraint\",\n",
      "      \"code_for_constraint\": \".isUnique(\\\"review_body\\\")\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"ComplianceConstraint(Compliance('marketplace' has value range '', 'FR', 'JP', 'UK', 'DE', 'US',`marketplace` IN ('', 'FR', 'JP', 'UK', 'DE', 'US'),None))\",\n",
      "      \"column_name\": \"marketplace\",\n",
      "      \"current_value\": \"Compliance: 1\",\n",
      "      \"description\": \"'marketplace' has value range '', 'FR', 'JP', 'UK', 'DE', 'US'\",\n",
      "      \"suggesting_rule\": \"CategoricalRangeRule(com.amazon.deequ.suggestions.rules.CategoricalRangeRule$$$Lambda$4119/0x000000080197e840@74f276b0)\",\n",
      "      \"rule_description\": \"If we see a categorical range for a column, we suggest an IS IN (...) constraint\",\n",
      "      \"code_for_constraint\": \".isContainedIn(\\\"marketplace\\\", [\\\"\\\", \\\"FR\\\", \\\"JP\\\", \\\"UK\\\", \\\"DE\\\", \\\"US\\\"])\"\n",
      "    },\n",
      "    {\n",
      "      \"constraint_name\": \"CompletenessConstraint(Completeness(marketplace,None))\",\n",
      "      \"column_name\": \"marketplace\",\n",
      "      \"current_value\": \"Completeness: 0.8570391886739565\",\n",
      "      \"description\": \"'marketplace' has less than 15% missing values\",\n",
      "      \"suggesting_rule\": \"RetainCompletenessRule()\",\n",
      "      \"rule_description\": \"If a column is incomplete in the sample, we model its completeness as a binomial variable, estimate a confidence interval and use this to define a lower bound for the completeness\",\n",
      "      \"code_for_constraint\": \".hasCompleteness(\\\"marketplace\\\", lambda x: x >= 0.85, \\\"It should be above 0.85!\\\")\"\n",
      "    }\n",
      "  ]\n",
      "}\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "from pydeequ.suggestions import *\n",
    "\n",
    "suggestionResult = ConstraintSuggestionRunner(spark) \\\n",
    "             .onData(df) \\\n",
    "             .addConstraintRule(DEFAULT()) \\\n",
    "             .run()\n",
    "\n",
    "# Constraint Suggestions in JSON format\n",
    "print(json.dumps(suggestionResult, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The result contains a list of constraints with descriptions and Python code, so that you can directly apply it in your data quality checks. You can call `print(json.dumps(result_json))` to inspect the suggested constraints."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Scaling to Production \n",
    "\n",
    "So far, we’ve shown you how to use these capabilities in the context of data exploration using a Jupyter notebook running on a SageMaker notebook instance. As your project matures, you need to use the same capabilities on larger and larger datasets, and in a production environment. With PyDeequ, it’s straightforward to make that transition. The following diagram illustrates deployment options for local and production purposes on AWS.\n",
    "\n",
    "![pydeequ-in-production](../imgs/pydeequ_deployment.png)\n",
    "\n",
    "**Figure 3. Deployment of PyDeequ in production.** \n",
    "\n",
    "As seen in the diagram above, you can leverage both an AWS EMR cluster and/or AWS Glue for larger or production purposes. To learn more about how to configure an EMR cluster with PyDeequ to explore much larger volumes of data please refer to the AWS blog post [Testing data quality at scale with PyDeequ](https://aws.amazon.com/blogs/big-data/testing-data-quality-at-scale-with-pydeequ/)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## More Examples on GitHub\n",
    "\n",
    "You can find examples of more advanced features on the [Deequ GitHub repo](https://github.com/awslabs/deequ):\n",
    "\n",
    "- Deequ provides more than data quality checks with fixed thresholds. Learn how to use [anomaly detection on data quality metrics](https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/anomaly_detection_example.md) to apply tests on metrics that change over time.\n",
    "- Deequ offers support for storing and loading metrics. Learn how to use the [MetricsRepository](https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/metrics_repository_example.md) for this use case.\n",
    "- If your dataset grows over time or is partitioned, you can use Deequ’s [incremental metrics computation](https://github.com/awslabs/deequ/blob/master/src/main/scala/com/amazon/deequ/examples/algebraic_states_example.md). For each partition, Deequ stores a state for each computed metric. To compute metrics for the union of partitions, Deequ can use these states to efficiently derive overall metrics without reloading the data.\n",
    "\n",
    "## Conclusion\n",
    "\n",
    "This notebook showed you how to use PyDeequ for calculating data quality metrics, verifying data quality metrics, and profiling data to automate the configuration of data quality checks in an Amazon SageMaker notebook. PyDeequ is available using `pip install` and on GitHub for you to build your own data quality management pipeline.\n",
    "\n",
    "Learn more about the inner workings of Deequ in the VLDB 2018 paper [Automating large-scale data quality verification](https://www.vldb.org/pvldb/vol11/p1781-schelter.pdf)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}