connect_and_validate_data.ipynb

{
    "cells": [
        {
            "cell_type": "markdown",
            "metadata": {
                "collapsed": true
            },
            "source": [
                "## Connection and Data Validation Notebook"
            ]
        },
        {
            "attachments": {},
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Table of Contents\n",
                "* [Check for Training Data in Project Space](#DataCheck)\n",
                "    * [Load the Training Data from COS if the file does not exist in the project space](#section_1_1)\n",
                "    \n",
                "    * [Check the connection and data loading](#section_1_2)\n",
                "  \n",
                "* [Data Validation](#chapter2)\n",
                "    * [Split the Data](#Optional)\n",
                "\n",
                "    * [Generate Training Stats on both Splits](#section_2_2)\n",
                "    * [Infer Schema on both Splits](#section_2_3) \n",
                "    * [Check for anomalies](#section_2_4) \n",
                "    * [Return a boolean to validate the tests](#section_3_1) \n"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Imports"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "from botocore.client import Config\n",
                "from sklearn.model_selection import train_test_split\n",
                "from dataclasses import dataclass\n",
                "import tensorflow_data_validation as tfdv\n",
                "import numpy as np\n",
                "import pandas as pd\n",
                "\n",
                "from ibm_watson_studio_pipelines import WSPipelines\n",
                "import ibm_boto3\n",
                "\n",
                "import logging\n",
                "import os, types\n",
                "import warnings\n",
                "\n",
                "warnings.filterwarnings(\"ignore\")"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "These environment variables are set in WS Pipelines"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "CLOUD_API_KEY = os.getenv(\"cloud_api_key\")\n",
                "training_file_name = os.getenv(\"training_file_name\")"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "### Load the Credentials\n"
            ]
        },
        {
            "attachments": {},
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "### Succeeding cell contains the credentials for MLOps COS. Please enter those before running the cell.\n",
                "```\n",
                "## PROJECT COS \n",
                "AUTH_ENDPOINT = \"https://iam.cloud.ibm.com/oidc/token\"\n",
                "ENDPOINT_URL = \"https://s3.private.us.cloud-object-storage.appdomain.cloud\"\n",
                "API_KEY_COS = \"xxx\"\n",
                "BUCKET_PROJECT_COS = \"mlops-donotdelete-pr-qxxcecxi1dtw94\"\n",
                "\n",
                "##MLOPS COS\n",
                "ENDPOINT_URL_MLOPS = \"https://s3.jp-tok.cloud-object-storage.appdomain.cloud\"\n",
                "API_KEY_MLOPS = \"xxx\"\n",
                "CRN_MLOPS = \"xxx\"\n",
                "BUCKET_MLOPS  = \"mlops-asset\"\n",
                "\n",
                "```\n",
                "\n",
                "**Alternatively, you may deliver your COS credentials for the project and the mlops-specific Bucket as a Global Pipeline Parameter (as a JSON String) within the Pipeline Canvas in Watson Studio.**"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "## Retrieve cos credentials from global pipeline parameters\n",
                "import json\n",
                "# Get json from environment and convert to string\n",
                "project_cos_credentials = json.loads(os.getenv('project_cos_credentials'))\n",
                "mlops_cos_credentials = json.loads(os.getenv('mlops_cos_credentials'))\n",
                "\n",
                "## PROJECT COS \n",
                "AUTH_ENDPOINT = project_cos_credentials['AUTH_ENDPOINT']\n",
                "ENDPOINT_URL = project_cos_credentials['ENDPOINT_URL']\n",
                "API_KEY_COS = project_cos_credentials['API_KEY']\n",
                "BUCKET_PROJECT_COS = project_cos_credentials['BUCKET']\n",
                "\n",
                "## MLOPS COS\n",
                "ENDPOINT_URL_MLOPS = mlops_cos_credentials['ENDPOINT_URL']\n",
                "API_KEY_MLOPS = mlops_cos_credentials['API_KEY']\n",
                "CRN_MLOPS = mlops_cos_credentials['CRN']\n",
                "BUCKET_MLOPS  = mlops_cos_credentials['BUCKET']"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Check for Training Data in Project Space"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "def check_for_file_in_project_cos(key):\n",
                "    \n",
                "    try:\n",
                "        def __iter__(self): return 0\n",
                "        mlops_client = ibm_boto3.client(service_name='s3',\n",
                "                                        ibm_api_key_id=API_KEY_COS,\n",
                "                                        ibm_auth_endpoint=AUTH_ENDPOINT,\n",
                "                                        config=Config(signature_version='oauth'),\n",
                "                                        endpoint_url=ENDPOINT_URL)\n",
                "\n",
                "        body = mlops_client.get_object(Bucket=BUCKET_PROJECT_COS,Key=key)['Body']\n",
                "        # add missing __iter__ method, so pandas accepts body as file-like object\n",
                "        if not hasattr(body, \"__iter__\"): body.__iter__ = types.MethodType( __iter__, body )\n",
                "        return body\n",
                "    except Exception as e :\n",
                "        print(e)\n",
                "        return False\n",
                "    \n",
                "    \n",
                "def read_data_from_mlops_cos(key):\n",
                "    def __iter__(self): return 0\n",
                "    MLOPS_DATA_STORE_client = ibm_boto3.client(\n",
                "        service_name='s3',\n",
                "        ibm_api_key_id=API_KEY_MLOPS,\n",
                "        ibm_service_instance_id=CRN_MLOPS,\n",
                "        ibm_auth_endpoint=AUTH_ENDPOINT,\n",
                "        config=Config(signature_version='oauth'),\n",
                "        endpoint_url=ENDPOINT_URL_MLOPS)\n",
                "\n",
                "    body = MLOPS_DATA_STORE_client.get_object(Bucket=BUCKET_MLOPS, Key=key)['Body']\n",
                "    # add missing __iter__ method, so pandas accepts body as file-like object\n",
                "    if not hasattr(body, \"__iter__\"): body.__iter__ = types.MethodType( __iter__, body )\n",
                "\n",
                "    gcf_df = pd.read_csv(body)\n",
                "    return gcf_df\n",
                "    \n",
                "    \n",
                "def load_data_from_project(key):\n",
                "    body = check_for_file_in_project_cos(key)\n",
                "    if body:\n",
                "        gcf_df = pd.read_csv(body)\n",
                "        return gcf_df\n",
                "    else:\n",
                "        print(\"\\n\")\n",
                "        print(f\"{key} file is probably not in project. Loading File from MLOps COS Bucket.\")\n",
                "        gcf_df = read_data_from_mlops_cos(key)\n",
                "        return gcf_df"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Load the Training Data from COS if the file doesn't exist"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "gcr_df = load_data_from_project(training_file_name)\n",
                "\n",
                "## Encode for ease of use with OpenScale\n",
                "gcr_df['Risk'] = gcr_df['Risk'].map({'Risk':1,'No Risk':0})\n",
                "gcr_df.head()"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Data Validation "
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "@dataclass\n",
                "class Datavalidation:\n",
                "    \"\"\"\n",
                "    \n",
                "    Data Validation Class\n",
                "    \n",
                "    \"\"\"\n",
                "    dataframe : pd.DataFrame\n",
                "    mask_per :int\n",
                "    \n",
                "    \n",
                "    def split_data(self,seed=32):\n",
                "        \"\"\"\n",
                "        Split Data into Train and Test Splits\n",
                "        \n",
                "        \"\"\"\n",
                "        np.random.seed(seed)\n",
                "        mask = np.random.rand(len(self.dataframe)) <= self.mask_per\n",
                "        training_data = gcr_df[mask]\n",
                "        testing_data = gcr_df[~mask]\n",
                "\n",
                "        print(f\"No. of training examples: {training_data.shape[0]}\")\n",
                "        print(f\"No. of testing examples: {testing_data.shape[0]}\")\n",
                "        \n",
                "        return training_data, testing_data\n",
                "    \n",
                "    \n",
                "    def save_data_in_cos(self,df,filename,key):\n",
                "        \"\"\"\n",
                "        \n",
                "        Save Data in IBM Cloud Object Storage\n",
                "        \n",
                "        \"\"\"\n",
                "        try:\n",
                "            df.to_csv(filename,index=False)\n",
                "            mlops_res = ibm_boto3.resource(\n",
                "                service_name='s3',\n",
                "                ibm_api_key_id=API_KEY_MLOPS,\n",
                "                ibm_service_instance_id=CRN_MLOPS,\n",
                "                ibm_auth_endpoint=AUTH_ENDPOINT,\n",
                "                config=Config(signature_version='oauth'),\n",
                "                endpoint_url=ENDPOINT_URL_MLOPS)\n",
                "\n",
                "            mlops_res.Bucket(BUCKET_MLOPS).upload_file(filename,key)\n",
                "            print(f\"File {filename} uploaded successfully\")\n",
                "        except Exception as e:\n",
                "            print(e)\n",
                "            print(\"File upload for {filename} failed\")\n",
                "    \n",
                "    \n",
                "    def generate_statistics(self,df):\n",
                "        \"\"\"\n",
                "        \n",
                "        Generate Statistics on a given Dataframe\n",
                "        \n",
                "        \"\"\"\n",
                "        train_stats = tfdv.generate_statistics_from_dataframe(df)\n",
                "        tfdv.visualize_statistics(train_stats)\n",
                "        return train_stats\n",
                "    \n",
                "    def inferSchema(self,stats):\n",
                "        \n",
                "        \"\"\"\n",
                "        InferSchema on a given Dataframe\n",
                "        \n",
                "        \"\"\"\n",
                "        schema = tfdv.infer_schema(statistics=stats)\n",
                "        tfdv.display_schema(schema=schema)\n",
                "        return schema\n",
                "    \n",
                "    def compare_statistics(self,lhs,rhs):\n",
                "        \"\"\"\n",
                "        \n",
                "        Compare Statistics between a test dataframe and reference Schema\n",
                "        \n",
                "        \"\"\"\n",
                "        # Compare evaluation data with training data\n",
                "        tfdv.visualize_statistics(lhs_statistics=lhs, rhs_statistics=rhs,\n",
                "                                  lhs_name='TEST_DATASET', rhs_name='TRAIN_DATASET')\n",
                "        \n",
                "        \n",
                "    def check_for_anomalies(self,testable_stats,ref_schema):\n",
                "        \"\"\"\n",
                "        \n",
                "        Check for any anomalies based on statistics and schema and values\n",
                "        \n",
                "        \"\"\"\n",
                "        anomalies = tfdv.validate_statistics(statistics=testable_stats, schema=ref_schema)\n",
                "        tfdv.display_anomalies(anomalies)\n",
                "        if len(anomalies.anomaly_info.items()) > 0:\n",
                "            logger.error(\"Anomalies found in dataset...\")\n",
                "            logger.error(str(self.anomalies.anomaly_info.items()))\n",
                "            return True\n",
                "        else:\n",
                "            return False\n",
                "\n",
                "def check_if_file_exists(filename):\n",
                "    mlops_client = ibm_boto3.client(\n",
                "        service_name='s3',\n",
                "        ibm_api_key_id=API_KEY_MLOPS,\n",
                "        ibm_service_instance_id=CRN_MLOPS,\n",
                "        ibm_auth_endpoint=AUTH_ENDPOINT,\n",
                "        config=Config(signature_version='oauth'),\n",
                "        endpoint_url=ENDPOINT_URL_MLOPS)\n",
                "    \n",
                "    for key in mlops_client.list_objects(Bucket=BUCKET_MLOPS)['Contents']:\n",
                "        files = key['Key']\n",
                "        if files == filename:\n",
                "            return True\n",
                "    return False"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "###  Split Data into Train and Eval Splits to Check for Consistency"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "classvalidate = Datavalidation(dataframe=gcr_df,mask_per=0.8) \n",
                "\n",
                "training_data, testing_data = classvalidate.split_data()\n"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Generate Training Stats on both Splits"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "train_stats = classvalidate.generate_statistics(training_data)"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "test_stats = classvalidate.generate_statistics(testing_data)"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Infer Training Data Schema"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "train_schema = classvalidate.inferSchema(train_stats)"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Infer Test Data Schema"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "test_schema = classvalidate.inferSchema(test_stats)"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Compare Eval and Train Data "
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "classvalidate.compare_statistics(lhs=test_stats,rhs=train_stats)"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Check For Data Anomalies "
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "### Check eval data for errors by validating the eval data stats using the previously inferred schema."
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "anomaly_status = classvalidate.check_for_anomalies(test_stats,train_schema)\n",
                "anomaly_status"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Save Train and Test Data for Data Preparation Stage"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "if not anomaly_status:\n",
                "    classvalidate.save_data_in_cos(df=training_data,filename=\"train_gcr.csv\",key=\"train_gcr.csv\")\n",
                "    classvalidate.save_data_in_cos(df=testing_data,filename=\"test_gcr.csv\",key=\"test_gcr.csv\")"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Check if files Exists in COS"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "files_copied_in_cos = check_if_file_exists(\"train_gcr.csv\") and check_if_file_exists(\"test_gcr.csv\")\n",
                "files_copied_in_cos"
            ]
        },
        {
            "cell_type": "markdown",
            "metadata": {},
            "source": [
                "## Register a Boolean Variable in WS Pipeline"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "validation_params = {}\n",
                "validation_params['anomaly_status'] = anomaly_status\n",
                "validation_params['files_copied_in_cos'] = files_copied_in_cos"
            ]
        },
        {
            "cell_type": "code",
            "execution_count": null,
            "metadata": {},
            "outputs": [],
            "source": [
                "pipelines_client = WSPipelines.from_apikey(apikey=CLOUD_API_KEY)\n",
                "pipelines_client.store_results(validation_params)"
            ]
        }
    ],
    "metadata": {
        "kernelspec": {
            "display_name": "Python 3",
            "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.10.10"
        },
        "vscode": {
            "interpreter": {
                "hash": "bd385fe162c5ca0c84973b7dd5c518456272446b2b64e67c2a69f949ca7a1754"
            }
        }
    },
    "nbformat": 4,
    "nbformat_minor": 1
}