GE_PythonStoredProc.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "76ed59e1-ef9e-4053-88f9-2fa17b72168a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(0, 7, 0)\n",
      "[Row(CURRENT_WAREHOUSE()='CLUSTER1', CURRENT_DATABASE()='NYCTAXI', CURRENT_SCHEMA()='TAXI')]\n"
     ]
    }
   ],
   "source": [
    "from snowflake.snowpark.session import Session\n",
    "from snowflake.snowpark.functions import udf, avg, col,lit,call_udf,countDistinct,sproc,udf\n",
    "from snowflake.snowpark.types import IntegerType, FloatType, StringType, BooleanType\n",
    "import pandas as pd\n",
    "from configs.config import snowflake_conn_prop_local as snowflake_conn_prop\n",
    "import sys\n",
    "import json\n",
    "import platform\n",
    "import os,requests\n",
    "from pathlib import Path\n",
    "import glob\n",
    "from src.DataValidationContext import *\n",
    "\n",
    "\n",
    "from snowflake.snowpark import version\n",
    "print(version.VERSION)\n",
    "session = Session.builder.configs(snowflake_conn_prop).create()\n",
    "\n",
    "print(session.sql('select current_warehouse(), current_database(), current_schema()').collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc13ace2-9ede-47db-bcce-42ec5eb55ea3",
   "metadata": {},
   "source": [
    "### Initialize\n",
    "\n",
    "- Here we are downloading the GE 15.14 version from the Pypi to theb local folder on the machine where this code is running.\n",
    "- This has been tested with GE 15.14. There was some issue with 15.17 where its trying to look for the package ipywidgets which is found in Snowflake anaconda channel but not found in information_schema.packages \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "59d618b1-3881-4f5d-8fb8-59e7a4c7483b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "./temp/tarfiles/great_expectations-0.15.14.tar.gz\n",
      "Create local dir: ./temp/tarfiles\n",
      "Create local dir: ./temp/libs\n",
      "Downloading library from PyPI to ./temp/tarfiles ...\n",
      "Started Extracting GE tar file to ./temp/tarfiles ...\n",
      "Done extracting GE tar file to ./temp/tarfiles ...\n"
     ]
    }
   ],
   "source": [
    "\n",
    "PROJECT_HOME_DIR = '.'\n",
    "LOCAL_TEMP_DIR = os.path.join(PROJECT_HOME_DIR, 'temp') \n",
    "LOCAL_LIB_DIR = os.path.join(LOCAL_TEMP_DIR, 'libs')\n",
    "LOCAL_TARFile_DIR = os.path.join(LOCAL_TEMP_DIR, 'tarfiles')\n",
    "LIB_URLS = [\n",
    "    #'https://files.pythonhosted.org/packages/9f/57/1539d783553f3d67cea1b55d7fe494373c5c0c9af689d4c0e0c2d3197739/great_expectations-0.15.17-py3-none-any.whl'\n",
    "    'https://files.pythonhosted.org/packages/8e/9d/cecb12289f7967b15facf550a0bbb9c1e910968c3a61b91fd8cdb80aeb3c/great_expectations-0.15.14.tar.gz'\n",
    "    \n",
    "]\n",
    "\n",
    "for lib_url in LIB_URLS:\n",
    "    # get the file name, from the url\n",
    "    splits = lib_url.split('/')\n",
    "    tot_splits = len(splits)\n",
    "    target_file = splits[-1]\n",
    "    \n",
    "    local_lib_fl = f'{LOCAL_TARFile_DIR}/{target_file}'\n",
    "    print(local_lib_fl)\n",
    "\n",
    "    # Create a local directory for TAR and extracting tar..\n",
    "    Path(LOCAL_TARFile_DIR).mkdir(parents=True, exist_ok=True)\n",
    "    print(f'Create local dir: {LOCAL_TARFile_DIR}')\n",
    "\n",
    "    Path(LOCAL_LIB_DIR).mkdir(parents=True, exist_ok=True)\n",
    "    print(f'Create local dir: {LOCAL_LIB_DIR}')\n",
    "\n",
    "    print(f'Downloading library from PyPI to {LOCAL_TARFile_DIR} ...')\n",
    "    with open(local_lib_fl, \"wb\") as f:\n",
    "        r = requests.get(lib_url)\n",
    "        f.write(r.content)\n",
    "\n",
    "        \n",
    "# Extract GE tar file\n",
    "\n",
    "import tarfile\n",
    "file = tarfile.open(local_lib_fl)\n",
    "print(f'Started Extracting GE tar file to {LOCAL_TARFile_DIR} ...')\n",
    "file.extractall(f'{LOCAL_LIB_DIR}/ge')\n",
    "file.close()\n",
    "print(f'Done extracting GE tar file to {LOCAL_TARFile_DIR} ...')\n",
    "                                    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "0d673dfa-e374-4ac4-b6ee-0dbcbaf36fae",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "./temp/libs/ge/great_expectations-0.15.14/great_expectations\n"
     ]
    }
   ],
   "source": [
    "# Getting the path for the great_expectation folder after the tar file is extracted.\n",
    "\n",
    "import glob\n",
    "ge_import_path=''\n",
    "for result in glob.iglob('./temp/libs/ge/great_expectations*'):\n",
    "    ge_import_path=result+'/great_expectations'\n",
    "print(ge_import_path)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c9a03bd0-f804-4e82-a531-63aa8b50eecf",
   "metadata": {},
   "source": [
    "### Creating Python Stored Procedure"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "70f8a07d-8c14-454b-b2cf-a7918c9c0590",
   "metadata": {},
   "source": [
    "#### - SP will read all the python files from the src folder which has all the code for creating context, batches, expectations and validations required for Great Expectations\n",
    "#### - SP takes in the below parameters\n",
    "        - datasourcename : This can be any name that uniquely identifies the datasource for the data\n",
    "        - expectationsuitename : Expecatation Suite Name which has all the expectations that you want to create.\n",
    "        - checkpointname : Name of GE checkpoint.\n",
    "        - sftablename : Name of the table you want to create in Snowflake which will store all the validation results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "caac0c0e-f54f-43b8-ae50-1022e2499b98",
   "metadata": {},
   "outputs": [],
   "source": [
    "from configs.config import snowflake_conn_prop_local as snowflake_conn_prop\n",
    "from src.DataValidationContext import GEDataValidationContext\n",
    "from src.BatchRequest import getBatchRequest \n",
    "from src.Expectations import  createExpectationSuite, createExpectations\n",
    "from src.RunLoadExpectations import runExpectaionValidation,loadValidationToDB\n",
    "\n",
    "import json\n",
    "import os\n",
    "\n",
    "\n",
    "session.sql(\"create or replace stage phani_greatexpectation\").collect()\n",
    "session.clear_packages()\n",
    "session.add_packages('pyparsing','pandas','pycryptodomex','boto3','tzlocal','tqdm','requests','ruamel.yaml','ipython','jsonpatch','mistune','jinja2','jsonschema','scipy','altair','Click','colorama','cryptography','snowflake-snowpark-python','sqlalchemy','chardet','asn1crypto')\n",
    "session.clear_imports()\n",
    "# session.add_import('great_expectations')\n",
    "session.add_import(ge_import_path)\n",
    "session.add_import('src')\n",
    "session.add_import('configs')\n",
    "\n",
    "@sproc(session=session,name=\"usp_generateGEValidationResults\", replace=True, return_type=StringType(),input_types=[StringType(),StringType(),StringType(),StringType()], is_permanent=True, stage_location='@phani_greatexpectation/ge_AllLibs')\n",
    "def generateGEValidationResults(session: Session,datasourcename:str,expecationsuitename:str,checkpointname:str,sftablename) -> str:\n",
    "    \n",
    "    from pathlib import Path\n",
    "    import os ,sys ,json ,tarfile\n",
    "    \n",
    "    #Creating GE context inside code\n",
    "    ge=GEDataValidationContext(datasourcename)\n",
    "    context=ge.getContext()\n",
    "    \n",
    "    # Creating the Pandas DataFrame from Snowpark DF\n",
    "    pd_df=session.sql(\"select top 2000 * from TAXI_TRIPS_MAT_VIEW\").to_pandas()\n",
    "    \n",
    "    # Getting the batch request used while creating and running validation on expectations\n",
    "    local_batch_request=getBatchRequest(context,datasourcename,pd_df)\n",
    "    \n",
    "    #Creating GE expectation Suite\n",
    "    createExpectationSuite(context,expecationsuitename)\n",
    "    \n",
    "    #Creating GE expecations\n",
    "    createExpectations(context,expecationsuitename,local_batch_request,pd_df)\n",
    "    \n",
    "    #Running GE validation \n",
    "    res=runExpectaionValidation(context,\"checkpointname\",local_batch_request,expecationsuitename,datasourcename)\n",
    "    \n",
    "    #Loading validation result to Snowflake table. Using append option while writing the data to the table\n",
    "    loadValidationToDB(session,res,sftablename)\n",
    "      \n",
    "    return 'SUCCESS'\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "4e3abb2f-dda0-4bae-93fa-60bcb7539abb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(USP_GENERATEGEVALIDATIONRESULTS='SUCCESS')]"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Calling SP\n",
    "session.sql(\"call usp_generateGEValidationResults('PandasDataSource','TaxiExpecatation','TaxiCheckpoint','TaxiGEValidation')\").collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "848e67d5-24c5-4f7c-9cfb-c62345dd5736",
   "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}