tutorials/data_profiling_example.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Single Column Profiling Example\n",
    "\n",
    "Very often we are faced with large, raw datasets and struggle to make sense of the data. A common example might be that we are given a huge CSV file and want to understand and clean the data contained therein. PyDeequ supports single-column profiling of such data and its implementation scales to large datasets with billions of rows. In the following, we showcase the basic usage of this profiling functionality:\n",
    "\n",
    "Assume we have raw data that is string typed (such as the data you would get from a CSV file). For the sake of simplicity, we use the following toy data in this example:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pydeequ\n",
    "\n",
    "import sagemaker_pyspark\n",
    "from pyspark.sql import SparkSession, Row\n",
    "\n",
    "classpath = \":\".join(sagemaker_pyspark.classpath_jars()) # aws-specific 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())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.sparkContext.parallelize([\n",
    "    Row(productName=\"thingA\", totalNumber=\"13.0\", status=\"IN_TRANSIT\", valuable=\"true\"),\n",
    "    Row(productName=\"thingA\", totalNumber=\"5\", status=\"DELAYED\", valuable=\"false\"),\n",
    "    Row(productName=\"thingB\", totalNumber=None, status=\"DELAYED\", valuable=None),\n",
    "    Row(productName=\"thingC\", totalNumber=None, status=\"IN_TRANSIT\", valuable=\"false\"),\n",
    "    Row(productName=\"thingD\", totalNumber=\"1.0\", status=\"DELAYED\", valuable=\"true\"),\n",
    "    Row(productName=\"thingC\", totalNumber=\"7.0\", status=\"UNKNOWN\", valuable=None),\n",
    "    Row(productName=\"thingC\", totalNumber=\"20\", status=\"UNKNOWN\", valuable=None),\n",
    "    Row(productName=\"thingE\", totalNumber=\"20\", status=\"DELAYED\", valuable=\"false\")]).toDF()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It only takes a single method invocation to make **PyDeequ** profile this data. Note that it will execute the three passes over the data and avoid any shuffles in order to easily scale to large data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydeequ.profiles import *\n",
    "\n",
    "result = ColumnProfilerRunner(spark) \\\n",
    "            .onData(df) \\\n",
    "            .run()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As a result, we get a profile for each column in the data, which allows us to inspect the completeness of the column, the approximate number of distinct values and the inferred datatype.\n",
    "\n",
    "In case of our toy data, we would get the following profiling results. Note that **PyDeequ** detected that `totalNumber` is a fractional column (and could be casted to float or double type) and that `valuable` is a boolean column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Column 'productName'\n",
      "\t completeness: 1.0\n",
      "\t approximate number of distinct values: 5\n",
      "\t datatype: String\n",
      "Column 'status'\n",
      "\t completeness: 1.0\n",
      "\t approximate number of distinct values: 3\n",
      "\t datatype: String\n",
      "Column 'totalNumber'\n",
      "\t completeness: 0.75\n",
      "\t approximate number of distinct values: 5\n",
      "\t datatype: Fractional\n",
      "Column 'valuable'\n",
      "\t completeness: 0.625\n",
      "\t approximate number of distinct values: 2\n",
      "\t datatype: Boolean\n"
     ]
    }
   ],
   "source": [
    "for col, profile in result.profiles.items():\n",
    "    print(f'Column \\'{col}\\'')\n",
    "    print('\\t',f'completeness: {profile.completeness}')\n",
    "    print('\\t',f'approximate number of distinct values: {profile.approximateNumDistinctValues}')\n",
    "    print('\\t',f'datatype: {profile.dataType}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For numeric columns, we get an extended profile which also contains descriptive statistics.\n",
    "\n",
    "For the `totalNumber` column we can inspect its minimum, maximum, mean and standard deviation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Statistics of 'totalNumber':\n",
      "\t minimum: 1.0\n",
      "\t maximum: 20.0\n",
      "\t mean: 11.0\n",
      "\t standard deviation: 7.280109889280518\n"
     ]
    }
   ],
   "source": [
    "totalNumber_profile = result.profiles['totalNumber']\n",
    "\n",
    "print(f'Statistics of \\'totalNumber\\':')\n",
    "print('\\t',f\"minimum: {totalNumber_profile.minimum}\")\n",
    "print('\\t',f\"maximum: {totalNumber_profile.maximum}\")\n",
    "print('\\t',f\"mean: {totalNumber_profile.mean}\")\n",
    "print('\\t',f\"standard deviation: {totalNumber_profile.stdDev}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For columns with a low number of distinct values, we collect the full value distribution. Here are accurate statistics about the values in the `status` column:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Value distribution in 'status':\n",
      "\t IN_TRANSIT occurred 2 times (ratio is 0.25)\n",
      "\t UNKNOWN occurred 2 times (ratio is 0.25)\n",
      "\t DELAYED occurred 4 times (ratio is 0.5)\n"
     ]
    }
   ],
   "source": [
    "status_profile = result.profiles['status']\n",
    "\n",
    "print('Value distribution in \\'status\\':')\n",
    "for unique_entry in status_profile.histogram: \n",
    "    print('\\t',f\"{unique_entry.value} occurred {unique_entry.count} times (ratio is {unique_entry.ratio})\")"
   ]
  }
 ],
 "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
}