tutorials/anomaly_detection.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Anomaly Detection Basic Tutorial\n",
    "\n",
    "This Jupyter notebook will give a basic tutorial on how to use PyDeequ's Anomaly Detection module.\n",
    "\n",
    "Often times in dealing with large datasets it is difficult to understand the constraints needed in our data. However, we often have a better understanding of how much change we expect in certain metrics of our data. Therefore, we can use anomaly detection to measure the data quality in large datasets. \n",
    "\n",
    "The idea is that we regularly store the metrics in a MetricsRepository. Once we do that, we can run anomaly checks using the Verification Suite to compare the current values of the metric to its past values in order to detect anomalous changes. \n",
    "\n",
    "In this simple example, we compute the size of a dataset every day to ensure sure that the size does not drastically change. The number of rows on a given day should not be more than double of what we have seen before. \n",
    "\n",
    "First import the proper imports, the pydeequ repository and run a sparksession for your test. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession, Row, DataFrame\n",
    "import pandas as pd\n",
    "import sagemaker_pyspark\n",
    "\n",
    "import pydeequ\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",
    "    .getOrCreate())\n",
    "sc = spark.sparkContext\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Initialize Metrics Repository"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us import `repository`  to access the `InMemoryMetricsRepository` class needed for anomoly detection, and create a metrics repository to store the metrics. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydeequ.repository import *\n",
    "metricsRepository = InMemoryMetricsRepository(spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Let us do an anomaly detection check using Metrics Repository!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will be using our ficticious datasets from yesterday which contains only two rows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "yesterdaysDataset = sc.parallelize([\n",
    "            Row(a=3, b=0,),\n",
    "            Row(a=3, b=5,)]).toDF()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will be demoing the `RelativeRateOfChangeStrategy` for detecting anomalies on the analyzer `Size()`. With the `maxRateIncrease` parameter set as 2.0, this will test for anomalies in the size of the data to ensure that it does not increase by more than 2x the saved amount. Alternatively, we can use other strategies or analyzers better suited for analyzing our data.\n",
    "\n",
    "The resulting metrics are stored using `useRepository` and `saveOrAppendResult` under a result key: `yesterdaysKey` with yesterday's timestamp. To run the `VerificationSuite` use the `run()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydeequ.verification import *\n",
    "\n",
    "yesterdaysKey = ResultKey(spark, ResultKey.current_milli_time() - 24 * 60 * 60 * 1000)\n",
    "\n",
    "prev_Result = VerificationSuite(spark).onData(yesterdaysDataset) \\\n",
    "    .useRepository(metricsRepository) \\\n",
    "    .saveOrAppendResult(yesterdaysKey) \\\n",
    "    .addAnomalyCheck(RelativeRateOfChangeStrategy(maxRateIncrease=2.0), Size()) \\\n",
    "    .run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, the ficticious data from today has five rows, the data size more than doubled and should raise an anomaly detection. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "todaysDataset = sc.parallelize([\n",
    "            Row(a=3,  b=0,),\n",
    "            Row(a=3,  b=5,),\n",
    "            Row(a=100,b=5,),\n",
    "            Row(a=2,  b=30,),\n",
    "            Row(a=10, b=5,)]).toDF()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Repeat the anomaly check using our metrics repository and verification suite."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "todaysKey = ResultKey(spark, ResultKey.current_milli_time())\n",
    "\n",
    "currResult = VerificationSuite(spark).onData(todaysDataset) \\\n",
    "    .useRepository(metricsRepository) \\\n",
    "    .saveOrAppendResult(todaysKey) \\\n",
    "    .addAnomalyCheck(RelativeRateOfChangeStrategy(maxRateIncrease=2.0), Size()) \\\n",
    "    .run()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Let us detect any anomalies!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now look at the status of the verification to see if an anomaly has been detected. If detected (which it should have) the contents of our metrics repository will be printed. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Anomaly detected in the Size() metric!\n",
      "+-------+--------+----+-----+-------------+\n",
      "| entity|instance|name|value| dataset_date|\n",
      "+-------+--------+----+-----+-------------+\n",
      "|Dataset|       *|Size|  5.0|1594942992829|\n",
      "|Dataset|       *|Size|  2.0|1594856587188|\n",
      "+-------+--------+----+-----+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "if (currResult.status != \"Success\"):\n",
    "    print(\"Anomaly detected in the Size() metric!\")\n",
    "    metricsRepository.load().forAnalyzers([Size()]).getSuccessMetricsAsDataFrame().show()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "pycharm": {
     "name": "#%% md\n"
    }
   },
   "source": [
    "### We see that an anomaly has been detected in the dataset and it is due to a data size increase from 2 to 5!\n",
    "\n",
    "### For more info ... look at full list of strategies available for anomaly_detection in `docs/anomaly_detection.md`"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_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.6.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}