predictive_maintenance_pipeline_datarest_endpoint_success.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Predictive Maintenance Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import time\n",
    "import boto3\n",
    "import string\n",
    "import sagemaker\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import awswrangler as wr\n",
    "\n",
    "from sagemaker.feature_store.feature_group import FeatureGroup\n",
    "from sagemaker.workflow.pipeline_context import PipelineSession"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the region name\n",
    "region = sagemaker.Session().boto_region_name\n",
    "print(\"Using AWS Region: {}\".format(region))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Setup required sessions\n",
    "boto3.setup_default_session(region_name = region)\n",
    "boto_session = boto3.Session(region_name = region)\n",
    "\n",
    "s3_client = boto3.client(\"s3\", region_name = region)\n",
    "\n",
    "sagemaker_boto_client = boto_session.client(\"sagemaker\")\n",
    "sagemaker_session = sagemaker.session.Session(\n",
    "    boto_session = boto_session, sagemaker_client = sagemaker_boto_client\n",
    ")\n",
    "account_id = boto3.client(\"sts\").get_caller_identity()[\"Account\"]\n",
    "sagemaker_role = sagemaker.get_execution_role()\n",
    "pipeline_session = PipelineSession()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sagemaker.get_execution_role()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# S3 bucket where all the data will be stored\n",
    "bucket = \"BUCKET-NAME\"\n",
    "prefix = \"mlops/predictive-maintenance\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Upload raw data to S3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client.upload_file(\n",
    "    Filename=\"datasets/PdM_telemetry.csv\", Bucket=bucket, Key=f\"{prefix}/data/raw/PdM_telemetry.csv\"\n",
    ")\n",
    "s3_client.upload_file(\n",
    "    Filename=\"datasets/PdM_errors.csv\", Bucket=bucket, Key=f\"{prefix}/data/raw/PdM_errors.csv\"\n",
    ")\n",
    "s3_client.upload_file(\n",
    "    Filename=\"datasets/PdM_maint.csv\", Bucket=bucket, Key=f\"{prefix}/data/raw/PdM_maint.csv\"\n",
    ")\n",
    "s3_client.upload_file(\n",
    "    Filename=\"datasets/PdM_failures.csv\", Bucket=bucket, Key=f\"{prefix}/data/raw/PdM_failures.csv\"\n",
    ")\n",
    "s3_client.upload_file(\n",
    "    Filename=\"datasets/PdM_machines.csv\", Bucket=bucket, Key=f\"{prefix}/data/raw/PdM_machines.csv\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define Parameters to Parametrize Pipeline Execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.parameters import (\n",
    "    ParameterInteger,\n",
    "    ParameterString,\n",
    "    ParameterFloat,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "processing_instance_count = ParameterInteger(name = \"ProcessingInstanceCount\", default_value = 1)\n",
    "instance_type = ParameterString(name = \"TrainingInstanceType\", default_value = \"ml.m5.xlarge\")\n",
    "\n",
    "model_approval_status = ParameterString(\n",
    "    name = \"ModelApprovalStatus\", default_value = \"PendingManualApproval\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define a Processing Step for Feature Engineering"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Preprocessing Script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting scripts/preprocessing.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile scripts/preprocessing.py\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import boto3\n",
    "from io import StringIO\n",
    "import awswrangler as wr\n",
    "\n",
    "base_dir = \"/opt/ml/processing\"\n",
    "bucket = \"BUCKET-NAME\"\n",
    "prefix = \"mlops/predictive-maintenance\"\n",
    "\n",
    "# Upload the preprocessed data to S3\n",
    "def upload_file_s3(df, name):\n",
    "    boto3.setup_default_session(region_name = \"us-east-1\")\n",
    "    s3_client = boto3.client(\"s3\", region_name = \"us-east-1\")\n",
    "    with StringIO() as csv_buffer:\n",
    "        df.to_csv(csv_buffer, index = False)\n",
    "\n",
    "        response = s3_client.put_object(\n",
    "            Bucket = bucket, Key = f\"{prefix}/data/preprocessed/{name}.csv\", Body = csv_buffer.getvalue()\n",
    "        )\n",
    "        status = response.get(\"ResponseMetadata\", {}).get(\"HTTPStatusCode\")\n",
    "\n",
    "        if status == 200:\n",
    "            print(f\"Successful S3 put_object response. Status - {status}\")\n",
    "        else:\n",
    "            print(f\"Unsuccessful S3 put_object response. Status - {status}\")\n",
    "\n",
    "# Convert to datetime datatype\n",
    "def datetime_datatype(df):\n",
    "    print(\"Converting to type datetime\")\n",
    "    df['datetime'] = pd.to_datetime(df['datetime'], format=\"%Y-%m-%d %H:%M:%S\")\n",
    "    return df\n",
    "\n",
    "\n",
    "# Convert to category datatype\n",
    "def category_datatype(df, column_name):\n",
    "    print(\"Converting to type category\")\n",
    "    df[column_name] = df[column_name].astype('category')\n",
    "    return df\n",
    "\n",
    "\n",
    "# Lag Features from Telemetry\n",
    "def telemetry_features(df):\n",
    "    df = datetime_datatype(df)\n",
    "    # Calculate mean values for telemetry features -- 3 hours rolling window\n",
    "    print(\"Calculate mean values for telemetry features -- 3 hours rolling window\")\n",
    "    temp = []\n",
    "    fields = ['volt', 'rotate', 'pressure', 'vibration']\n",
    "    for col in fields:\n",
    "        temp.append(pd.pivot_table(df,\n",
    "                                   index = 'datetime',\n",
    "                                   columns = 'machineID',\n",
    "                                   values = col).resample('3H', closed = 'left', label = 'right').mean().unstack())\n",
    "    telemetry_mean_3h = pd.concat(temp, axis = 1)\n",
    "    telemetry_mean_3h.columns = [i + 'mean_3h' for i in fields]\n",
    "    telemetry_mean_3h.reset_index(inplace = True)\n",
    "\n",
    "    # repeat for standard deviation\n",
    "    print(\"Calculate standard deviation for telemetry features -- 3 hours rolling window\")\n",
    "    temp = []\n",
    "    for col in fields:\n",
    "        temp.append(pd.pivot_table(df,\n",
    "                                   index = 'datetime',\n",
    "                                   columns = 'machineID',\n",
    "                                   values = col).resample('3H', closed = 'left', label = 'right').std().unstack())\n",
    "    telemetry_sd_3h = pd.concat(temp, axis = 1)\n",
    "    telemetry_sd_3h.columns = [i + 'sd_3h' for i in fields]\n",
    "    telemetry_sd_3h.reset_index(inplace = True)\n",
    "    \n",
    "    # Calculate mean values for telemetry features -- 24 hours rolling window\n",
    "    print(\"Calculate mean values for telemetry features -- 24 hours rolling window\")\n",
    "    temp = []\n",
    "    fields = ['volt', 'rotate', 'pressure', 'vibration']\n",
    "    for col in fields:\n",
    "        temp.append(pd.pivot_table(df,\n",
    "                                   index = 'datetime',\n",
    "                                   columns = 'machineID',\n",
    "                                   values = col)\n",
    "                    .resample('3H', closed = 'left', label = 'right')\n",
    "                    .first()\n",
    "                    .unstack()\n",
    "                    .rolling(window = 24, center = False).mean())\n",
    "    telemetry_mean_24h = pd.concat(temp, axis = 1)\n",
    "    telemetry_mean_24h.columns = [i + 'mean_24h' for i in fields]\n",
    "    telemetry_mean_24h.reset_index(inplace = True)\n",
    "    telemetry_mean_24h = telemetry_mean_24h.loc[-telemetry_mean_24h['voltmean_24h'].isnull()]\n",
    "\n",
    "    # repeat for standard deviation\n",
    "    print(\"Calculate standard deviation for telemetry features -- 24 hours rolling window\")\n",
    "    temp = []\n",
    "    fields = ['volt', 'rotate', 'pressure', 'vibration']\n",
    "    for col in fields:\n",
    "        temp.append(pd.pivot_table(df,\n",
    "                                   index = 'datetime',\n",
    "                                   columns = 'machineID',\n",
    "                                   values = col)\n",
    "                    .resample('3H', closed='left', label='right')\n",
    "                    .first()\n",
    "                    .unstack()\n",
    "                    .rolling(window = 24, center = False).std())\n",
    "    telemetry_sd_24h = pd.concat(temp, axis = 1)\n",
    "    telemetry_sd_24h.columns = [i + 'sd_24h' for i in fields]\n",
    "    telemetry_sd_24h = telemetry_sd_24h.loc[-telemetry_sd_24h['voltsd_24h'].isnull()]\n",
    "    telemetry_sd_24h.reset_index(inplace = True)\n",
    "    \n",
    "    telemetry_feat = pd.concat([telemetry_mean_3h,\n",
    "                            telemetry_sd_3h.iloc[:, 2:6],\n",
    "                            telemetry_mean_24h.iloc[:, 2:6],\n",
    "                            telemetry_sd_24h.iloc[:, 2:6]], axis = 1).dropna()\n",
    "\n",
    "    upload_file_s3(telemetry_feat, \"telemetry\")\n",
    "    \n",
    "    return telemetry_feat\n",
    "\n",
    "\n",
    "# Lag Features for Errors\n",
    "def errors_lag_features(df):\n",
    "    df = datetime_datatype(df)\n",
    "    df = category_datatype(df, 'errorID')\n",
    "    print(\"Lag features for errors\")\n",
    "    error_count = pd.get_dummies(df.set_index('datetime')).reset_index()\n",
    "    error_count.columns = ['datetime', 'machineID', 'error1', 'error2', 'error3', 'error4', 'error5']\n",
    "    error_count = error_count.groupby(['machineID', 'datetime']).sum().reset_index()\n",
    "    error_count = telemetry[['datetime', 'machineID']].merge(error_count, on = ['machineID', 'datetime'], how = 'left').fillna(0.0)\n",
    "    temp = []\n",
    "    fields = ['error%d' % i for i in range(1, 6)]\n",
    "    for col in fields:\n",
    "        temp.append(pd.pivot_table(error_count,\n",
    "                                   index = 'datetime',\n",
    "                                   columns = 'machineID',\n",
    "                                   values = col)\n",
    "                    .resample('3H', closed='left', label='right')\n",
    "                    .first()\n",
    "                    .unstack()\n",
    "                    .rolling(window = 24, center = False).sum())\n",
    "    error_count = pd.concat(temp, axis = 1)\n",
    "    error_count.columns = [i + 'count' for i in fields]\n",
    "    error_count.reset_index(inplace = True)\n",
    "    error_count = error_count.dropna()\n",
    "    \n",
    "    upload_file_s3(error_count, \"errors\")\n",
    "    \n",
    "    return error_count\n",
    "\n",
    "\n",
    "# Maintenance Features\n",
    "def maintenance_features(df):\n",
    "    df = datetime_datatype(df)\n",
    "    df = category_datatype(df, 'comp')\n",
    "    print(\"Maintenance Features -- Days since last replacement\")\n",
    "    comp_rep = pd.get_dummies(df.set_index('datetime')).reset_index()\n",
    "    comp_rep.columns = ['datetime', 'machineID', 'comp1', 'comp2', 'comp3', 'comp4']\n",
    "\n",
    "    # combine repairs for a given machine in a given hour\n",
    "    comp_rep = comp_rep.groupby(['machineID', 'datetime']).sum().reset_index()\n",
    "\n",
    "    # add timepoints where no components were replaced\n",
    "    comp_rep = telemetry[['datetime', 'machineID']].merge(comp_rep,\n",
    "                                                          on=['datetime', 'machineID'],\n",
    "                                                          how='outer').fillna(0).sort_values(by=['machineID', 'datetime'])\n",
    "    components = ['comp1', 'comp2', 'comp3', 'comp4']\n",
    "    for comp in components:\n",
    "        comp_rep.loc[comp_rep[comp] < 1, comp] = None\n",
    "        comp_rep.loc[-comp_rep[comp].isnull(),\n",
    "                     comp] = comp_rep.loc[-comp_rep[comp].isnull(), 'datetime']\n",
    "        comp_rep[comp] = comp_rep[comp].fillna(method = 'ffill')\n",
    "\n",
    "    comp_rep = comp_rep.loc[comp_rep['datetime'] > pd.to_datetime('2015-01-01')]\n",
    "    for comp in components:\n",
    "        comp_rep[comp] = (comp_rep[\"datetime\"] - pd.to_datetime(comp_rep[comp])) / np.timedelta64(1, \"D\")\n",
    "        \n",
    "    upload_file_s3(comp_rep, \"maint\")\n",
    "    \n",
    "    return comp_rep\n",
    "\n",
    "\n",
    "# Failures Features\n",
    "def failure_features(df):\n",
    "    print(\"Failure features\")\n",
    "    df = datetime_datatype(df)\n",
    "    df = category_datatype(df, 'failure')\n",
    "    upload_file_s3(df, \"failures\")\n",
    "    return df\n",
    "\n",
    "\n",
    "# Final Features\n",
    "def final_features(telemetry_df, errors_df, maint_df, machines_df):\n",
    "    upload_file_s3(machines_df, \"machines\")\n",
    "    print(\"Final features\")\n",
    "    final_feat = telemetry_df.merge(errors_df, on = ['datetime', 'machineID'], how = 'left')\n",
    "    final_feat = final_feat.merge(maint_df, on = ['datetime', 'machineID'], how = 'left')\n",
    "    final_feat = final_feat.merge(machines_df, on = ['machineID'], how = 'left')\n",
    "    return final_feat\n",
    "\n",
    "\n",
    "# Label Construction\n",
    "def label_construct(tele_df, error_df, maint_df, machine_df, failure_df):\n",
    "    print(\"----- Final Features -----\")\n",
    "    final_feat = final_features(tele_df, error_df, maint_df, machine_df)\n",
    "    \n",
    "    print(\"----- Label Construction -----\")\n",
    "    labeled_features = pd.DataFrame()\n",
    "    labeled_features = final_feat.merge(\n",
    "        failure_df, on = ['datetime', 'machineID'], how = 'left')\n",
    "    labeled_features['failure'] = labeled_features['failure'].astype(str)\n",
    "    labeled_features['failure'] = labeled_features['failure'].fillna(method = 'bfill', limit = 7)\n",
    "    labeled_features['failure'] = labeled_features['failure'].replace('nan', 'none')\n",
    "    print(\"----- Preprocessing completed -----\")\n",
    "    \n",
    "    upload_file_s3(labeled_features, \"preprocessed\")\n",
    "#     pd.DataFrame(labeled_features).to_csv(f\"{base_dir}/preprocessed/final_data.csv\", index = False)\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "\n",
    "    telemetry_data_uri = f\"s3://{bucket}/{prefix}/data/raw/PdM_telemetry.csv\"\n",
    "    errors_data_uri = f\"s3://{bucket}/{prefix}/data/raw/PdM_errors.csv\"\n",
    "    maint_data_uri = f\"s3://{bucket}/{prefix}/data/raw/PdM_maint.csv\"\n",
    "    failures_data_uri = f\"s3://{bucket}/{prefix}/data/raw/PdM_failures.csv\"\n",
    "    machines_data_uri = f\"s3://{bucket}/{prefix}/data/raw/PdM_machines.csv\"\n",
    "    \n",
    "    telemetry = wr.s3.read_csv(telemetry_data_uri)\n",
    "    errors = wr.s3.read_csv(errors_data_uri)\n",
    "    maint = wr.s3.read_csv(maint_data_uri)\n",
    "    failures = wr.s3.read_csv(failures_data_uri)\n",
    "    machines = wr.s3.read_csv(machines_data_uri)\n",
    "    \n",
    "    telemetry_df = telemetry_features(telemetry)\n",
    "    errors_df = errors_lag_features(errors)\n",
    "    maint_df = maintenance_features(maint)\n",
    "    failures_df = failure_features(failures)\n",
    "    machines_df = category_datatype(machines, 'model')\n",
    "    \n",
    "    label_construct(telemetry_df, errors_df, maint_df, machines_df, failures_df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Feature Store Creation Script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting scripts/featurestore.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile scripts/featurestore.py\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import boto3\n",
    "import sagemaker\n",
    "from sagemaker.feature_store.feature_group import FeatureGroup\n",
    "import awswrangler as wr\n",
    "\n",
    "base_dir = \"/opt/ml/processing\"\n",
    "bucket = \"BUCKET-NAME\"\n",
    "prefix = \"mlops/predictive-maintenance\"\n",
    "\n",
    "boto_session = boto3.Session(region_name = \"us-east-1\")\n",
    "sagemaker_boto_client = boto_session.client(\"sagemaker\")\n",
    "featurestore_runtime = boto_session.client(\n",
    "    service_name = \"sagemaker-featurestore-runtime\", region_name = \"us-east-1\"\n",
    ")\n",
    "try:\n",
    "    sagemaker_role = sagemaker.get_execution_role()\n",
    "    print(f\"Sagemaker Role for Feature Store file: {sagemaker_role}\")\n",
    "except ValueError:\n",
    "    sagemaker_role = 'SAGEMAKER-ROLE'\n",
    "    \n",
    "feature_store_session = sagemaker.Session(\n",
    "    boto_session = boto_session,\n",
    "    sagemaker_client = sagemaker_boto_client,\n",
    "    sagemaker_featurestore_runtime_client = featurestore_runtime,\n",
    ")\n",
    "\n",
    "# ------------------------------------ Read Data\n",
    "telemetry_data_uri = f\"s3://{bucket}/{prefix}/data/preprocessed/telemetry.csv\"\n",
    "errors_data_uri = f\"s3://{bucket}/{prefix}/data/preprocessed/errors.csv\"\n",
    "maint_data_uri = f\"s3://{bucket}/{prefix}/data/preprocessed/maint.csv\"\n",
    "failures_data_uri = f\"s3://{bucket}/{prefix}/data/preprocessed/failures.csv\"\n",
    "machines_data_uri = f\"s3://{bucket}/{prefix}/data/preprocessed/machines.csv\"\n",
    "\n",
    "telemetry = wr.s3.read_csv(telemetry_data_uri)\n",
    "errors = wr.s3.read_csv(errors_data_uri)\n",
    "maint = wr.s3.read_csv(maint_data_uri)\n",
    "failures = wr.s3.read_csv(failures_data_uri)\n",
    "machines = wr.s3.read_csv(machines_data_uri)\n",
    "\n",
    "# ------------------------------------ Add Timestamp\n",
    "telemetry['event_time'] = pd.to_datetime(\"now\").timestamp()\n",
    "errors['event_time'] = pd.to_datetime(\"now\").timestamp()\n",
    "maint['event_time'] = pd.to_datetime(\"now\").timestamp()\n",
    "failures['event_time'] = pd.to_datetime(\"now\").timestamp()\n",
    "machines['event_time'] = pd.to_datetime(\"now\").timestamp()\n",
    "\n",
    "# ------------------------------------ Create Feature Group\n",
    "telemetry_feature_group = FeatureGroup(name = 'telemetry_fg', sagemaker_session = feature_store_session)\n",
    "errors_feature_group = FeatureGroup(name = 'errors_fg', sagemaker_session = feature_store_session)\n",
    "maintenance_feature_group = FeatureGroup(name = 'maintenance_fg', sagemaker_session = feature_store_session)\n",
    "failures_feature_group = FeatureGroup(name = 'failures_fg', sagemaker_session = feature_store_session)\n",
    "machines_feature_group = FeatureGroup(name = 'machines_fg', sagemaker_session = feature_store_session)\n",
    "\n",
    "# ------------------------------------ Loading Definitions\n",
    "telemetry_feature_group.load_feature_definitions(data_frame = telemetry)\n",
    "errors_feature_group.load_feature_definitions(data_frame = errors)\n",
    "maintenance_feature_group.load_feature_definitions(data_frame = maint)\n",
    "failures_feature_group.load_feature_definitions(data_frame = failures)\n",
    "machines_feature_group.load_feature_definitions(data_frame = machines)\n",
    "\n",
    "record_identifier_feature_name = \"machineID\"\n",
    "event_time_feature_name = \"event_time\"\n",
    "\n",
    "# ------------------------------------ Telemetry Feature Store\n",
    "try:\n",
    "    telemetry_feature_group.create(\n",
    "        s3_uri = f\"s3://{bucket}/{prefix}/feature_store_data\",\n",
    "        record_identifier_name = record_identifier_feature_name,\n",
    "        event_time_feature_name = event_time_feature_name,\n",
    "        role_arn = sagemaker_role,\n",
    "        enable_online_store = True,\n",
    "    )\n",
    "    time.sleep(30)\n",
    "    print(f'Create \"telemetry\" feature group: SUCCESS')\n",
    "except Exception as e:\n",
    "    code = e.response.get(\"Error\").get(\"Code\")\n",
    "    if code == \"ResourceInUse\":\n",
    "        print(\"Using existing feature group\")\n",
    "    else:\n",
    "        raise (e)\n",
    "\n",
    "# ------------------------------------ Errors Feature Store      \n",
    "try:\n",
    "    errors_feature_group.create(\n",
    "        s3_uri = f\"s3://{bucket}/{prefix}/feature_store_data\",\n",
    "        record_identifier_name = record_identifier_feature_name,\n",
    "        event_time_feature_name = event_time_feature_name,\n",
    "        role_arn = sagemaker_role,\n",
    "        enable_online_store = True,\n",
    "    )\n",
    "    time.sleep(30)\n",
    "    print(f'Create \"errors\" feature group: SUCCESS')\n",
    "except Exception as e:\n",
    "    code = e.response.get(\"Error\").get(\"Code\")\n",
    "    if code == \"ResourceInUse\":\n",
    "        print(\"Using existing feature group\")\n",
    "    else:\n",
    "        raise (e)\n",
    "\n",
    "# ------------------------------------ Maintenance Feature Store\n",
    "try:\n",
    "    maintenance_feature_group.create(\n",
    "        s3_uri = f\"s3://{bucket}/{prefix}/feature_store_data\",\n",
    "        record_identifier_name = record_identifier_feature_name,\n",
    "        event_time_feature_name = event_time_feature_name,\n",
    "        role_arn = sagemaker_role,\n",
    "        enable_online_store = True,\n",
    "    )\n",
    "    time.sleep(30)\n",
    "    print(f'Create \"maintenance\" feature group: SUCCESS')\n",
    "except Exception as e:\n",
    "    code = e.response.get(\"Error\").get(\"Code\")\n",
    "    if code == \"ResourceInUse\":\n",
    "        print(\"Using existing feature group\")\n",
    "    else:\n",
    "        raise (e)\n",
    "\n",
    "# ------------------------------------ Failures Feature Store\n",
    "try:\n",
    "    failures_feature_group.create(\n",
    "        s3_uri = f\"s3://{bucket}/{prefix}/feature_store_data\",\n",
    "        record_identifier_name = record_identifier_feature_name,\n",
    "        event_time_feature_name = event_time_feature_name,\n",
    "        role_arn = sagemaker_role,\n",
    "        enable_online_store = True,\n",
    "    )\n",
    "    time.sleep(30)\n",
    "    print(f'Create \"failures\" feature group: SUCCESS')\n",
    "except Exception as e:\n",
    "    code = e.response.get(\"Error\").get(\"Code\")\n",
    "    if code == \"ResourceInUse\":\n",
    "        print(\"Using existing feature group\")\n",
    "    else:\n",
    "        raise (e)\n",
    "\n",
    "# ------------------------------------ Machines Feature Store\n",
    "try:\n",
    "    machines_feature_group.create(\n",
    "        s3_uri = f\"s3://{bucket}/{prefix}/feature_store_data\",\n",
    "        record_identifier_name = record_identifier_feature_name,\n",
    "        event_time_feature_name = event_time_feature_name,\n",
    "        role_arn = sagemaker_role,\n",
    "        enable_online_store = True,\n",
    "    )\n",
    "    time.sleep(30)\n",
    "    print(f'Create \"machines\" feature group: SUCCESS')\n",
    "except Exception as e:\n",
    "    code = e.response.get(\"Error\").get(\"Code\")\n",
    "    if code == \"ResourceInUse\":\n",
    "        print(\"Using existing feature group\")\n",
    "    else:\n",
    "        raise (e)\n",
    "\n",
    "# ------------------------------------ Ingesting Data\n",
    "while (\n",
    "    telemetry_feature_group.describe()['FeatureGroupStatus'] == 'Creating'):\n",
    "    print(\"Feature Group Creating\")\n",
    "    time.sleep(60)\n",
    "else:\n",
    "    print(\"Feature Group Created\")\n",
    "    ## Below code needs to run only once to ingest the data into feature store\n",
    "    #---------------------------------------------------------------\n",
    "#     telemetry_feature_group.ingest(data_frame = telemetry, max_workers = 3, wait = True)\n",
    "#     errors_feature_group.ingest(data_frame = errors, max_workers = 3, wait = True)\n",
    "#     maintenance_feature_group.ingest(data_frame = maint, max_workers = 3, wait = True)\n",
    "#     failures_feature_group.ingest(data_frame = failures, max_workers = 3, wait = True)\n",
    "#     machines_feature_group.ingest(data_frame = machines, max_workers = 3, wait = True)\n",
    "#     print(\"Feature Data Ingested\")\n",
    "    #---------------------------------------------------------------"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Train Test Split Script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting scripts/train_test_split_data.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile scripts/train_test_split_data.py\n",
    "import pandas as pd\n",
    "import boto3\n",
    "from io import StringIO\n",
    "import awswrangler as wr\n",
    "\n",
    "base_dir = \"/opt/ml/processing\"\n",
    "bucket = \"BUCKET-NAME\"\n",
    "prefix = \"mlops/predictive-maintenance\"\n",
    "\n",
    "def upload_file_s3(df, name):\n",
    "    boto3.setup_default_session(region_name = \"us-east-1\")\n",
    "    s3_client = boto3.client(\"s3\", region_name = \"us-east-1\")\n",
    "    with StringIO() as csv_buffer:\n",
    "        df.to_csv(csv_buffer, index = False)\n",
    "\n",
    "        response = s3_client.put_object(\n",
    "            Bucket = bucket, Key = f\"{prefix}/data/train-test/{name}.csv\", Body = csv_buffer.getvalue()\n",
    "        )\n",
    "        status = response.get(\"ResponseMetadata\", {}).get(\"HTTPStatusCode\")\n",
    "\n",
    "        if status == 200:\n",
    "            print(f\"Successful S3 put_object response. Status - {status}\")\n",
    "        else:\n",
    "            print(f\"Unsuccessful S3 put_object response. Status - {status}\")\n",
    "\n",
    "def train_test_split_script(labeled_features):\n",
    "    threshold_dates = [[pd.to_datetime('2015-07-31 01:00:00'), pd.to_datetime('2015-08-01 01:00:00')],\n",
    "                   [pd.to_datetime('2015-08-31 01:00:00'), pd.to_datetime('2015-09-01 01:00:00')],\n",
    "                   [pd.to_datetime('2015-09-30 01:00:00'), pd.to_datetime('2015-10-01 01:00:00')]]\n",
    "    \n",
    "    for last_train_date, first_test_date in threshold_dates:\n",
    "        # split out training and test data\n",
    "        print(labeled_features['datetime'][0])\n",
    "        train_y = labeled_features.loc[labeled_features['datetime'] < last_train_date, 'failure']\n",
    "        train_data = pd.get_dummies(labeled_features.loc[labeled_features['datetime'] < last_train_date].drop(['datetime',\n",
    "                                                                                                            'machineID',\n",
    "                                                                                                              'failure'], axis = 1))\n",
    "        test_y = labeled_features.loc[labeled_features['datetime'] > last_train_date, 'failure']\n",
    "        test_data = pd.get_dummies(labeled_features.loc[labeled_features['datetime'] > first_test_date].drop(['datetime',\n",
    "                                                                                                           'machineID',\n",
    "                                                                                                             'failure'], axis = 1))\n",
    "    \n",
    "    train_data['failure'] = train_y\n",
    "    test_data['failure'] = test_y\n",
    "    \n",
    "    upload_file_s3(train_data, \"train\")\n",
    "    upload_file_s3(test_data, \"test\")\n",
    "\n",
    "    pd.DataFrame(train_data).to_csv(f\"{base_dir}/train/train.csv\", index = False)\n",
    "    pd.DataFrame(test_data).to_csv(f\"{base_dir}/test/test.csv\", index = False)\n",
    "    \n",
    "if __name__ == \"__main__\":\n",
    "    final_data_uri = f\"s3://{bucket}/{prefix}/data/preprocessed/preprocessed.csv\"\n",
    "    final_data = wr.s3.read_csv(final_data_uri)\n",
    "    final_data['datetime'] = pd.to_datetime(final_data['datetime'], format=\"%Y-%m-%d %H:%M:%S\")\n",
    "    train_test_split_script(final_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an instance of a FrameworkProcessor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.processing import FrameworkProcessor\n",
    "from sagemaker.sklearn import SKLearn\n",
    "\n",
    "# Handles processing tasks for jobs using a machine learning framework\n",
    "sklearn_processor = FrameworkProcessor(\n",
    "    estimator_cls = SKLearn,\n",
    "    framework_version = \"1.2-1\",\n",
    "    instance_type = \"ml.m5.xlarge\",\n",
    "    instance_count = processing_instance_count,\n",
    "    base_job_name = \"sklearn-pred-maint-process\",\n",
    "    role = sagemaker_role,\n",
    "    sagemaker_session = pipeline_session,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Creating Processing Step"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/conda/lib/python3.7/site-packages/sagemaker/workflow/pipeline_context.py:261: UserWarning: Running within a PipelineSession, there will be No Wait, No Logs, and No Job being started.\n",
      "  UserWarning,\n"
     ]
    }
   ],
   "source": [
    "from sagemaker.processing import ProcessingInput, ProcessingOutput\n",
    "from sagemaker.workflow.steps import ProcessingStep\n",
    "\n",
    "# Runs a processing job\n",
    "processor_args = sklearn_processor.run(\n",
    "    code = \"preprocessing.py\",\n",
    "    source_dir = \"scripts\"\n",
    ")\n",
    "\n",
    "# Creates the pipeline step\n",
    "preprocess_step = ProcessingStep(name = \"PdM-Data-Read-And-PreProcessing\", step_args = processor_args)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Runs a processing job\n",
    "train_test_args = sklearn_processor.run(\n",
    "    outputs = [\n",
    "        ProcessingOutput(output_name = \"train\", source = \"/opt/ml/processing/train\"),\n",
    "        ProcessingOutput(output_name = \"test\", source = \"/opt/ml/processing/test\"),\n",
    "    ],\n",
    "    code = \"train_test_split_data.py\",\n",
    "    source_dir = \"scripts\"\n",
    ")\n",
    "\n",
    "# Creates the pipeline step\n",
    "train_test_split_step = ProcessingStep(name = \"PdM-Train-Test-Data-Split\", step_args = train_test_args, depends_on = [preprocess_step.name])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Runs a processing job\n",
    "fs_data = sklearn_processor.run(\n",
    "    code = \"featurestore.py\",\n",
    "    source_dir = \"scripts\"\n",
    ")\n",
    "\n",
    "# Creates the pipeline step\n",
    "feature_store_step = ProcessingStep(name = \"PdM-FeatureStore-Creation\", step_args = fs_data, depends_on = [preprocess_step.name])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define a Training Step to Train a Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.inputs import TrainingInput\n",
    "from sagemaker.workflow.steps import TrainingStep\n",
    "from sagemaker.sklearn.model import SKLearnModel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# SKLearn estimator is used for end-to-end training and deployment\n",
    "sklearn_estimator = SKLearn(\n",
    "    entry_point = \"scripts/rf_script-no-featurenames.py\",\n",
    "    role = sagemaker_role,\n",
    "    instance_count = 1,\n",
    "    instance_type = \"ml.m5.xlarge\",\n",
    "    framework_version = \"1.2-1\",\n",
    "    base_job_name = \"rf-scikit\",\n",
    "    hyperparameters = {\n",
    "        \"n-estimators\": 100,\n",
    "        \"min-samples-leaf\": 3,\n",
    "    },\n",
    "    sagemaker_session = pipeline_session,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_args = sklearn_estimator.fit(\n",
    "    inputs = {\n",
    "        \"train\": TrainingInput(\n",
    "            s3_data = train_test_split_step.properties.ProcessingOutputConfig.Outputs[\"train\"].S3Output.S3Uri,\n",
    "            content_type = \"text/csv\"\n",
    "        ),\n",
    "        \"test\": TrainingInput(\n",
    "            s3_data = train_test_split_step.properties.ProcessingOutputConfig.Outputs[\"test\"].S3Output.S3Uri,\n",
    "            content_type = \"text/csv\",\n",
    "        )\n",
    "    }\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Creates the pipeline step\n",
    "training_step = TrainingStep(\n",
    "    name = \"PdM-ModelTraininig\",\n",
    "    step_args = train_args,\n",
    "    depends_on = [train_test_split_step.name]\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define a Model Step to Create a Model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define a Register Model Step to Create a Model Package"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.step_collections import RegisterModel\n",
    "\n",
    "# Creates the pipeline step to register model\n",
    "register_step = RegisterModel(\n",
    "    name = \"PdM-Register-Model\",\n",
    "    estimator = sklearn_estimator,\n",
    "    model_data = training_step.properties.ModelArtifacts.S3ModelArtifacts,\n",
    "    content_types = [\"text/csv\"],\n",
    "    response_types = [\"text/csv\"],\n",
    "    inference_instances = [\"ml.t2.medium\", \"ml.m5.xlarge\"],\n",
    "    transform_instances = [\"ml.m5.xlarge\"],\n",
    "    model_package_group_name = \"predictive-maintenance\",\n",
    "    approval_status = model_approval_status,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 179,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/opt/conda/lib/python3.7/site-packages/sagemaker/workflow/pipeline_context.py:261: UserWarning: Running within a PipelineSession, there will be No Wait, No Logs, and No Job being started.\n",
      "  UserWarning,\n",
      "/opt/conda/lib/python3.7/site-packages/sagemaker/workflow/pipeline_context.py:261: UserWarning: Running within a PipelineSession, there will be No Wait, No Logs, and No Job being started.\n",
      "  UserWarning,\n"
     ]
    }
   ],
   "source": [
    "deploy_args = sklearn_processor.run(\n",
    "    code = \"deploy_model.py\",\n",
    "    source_dir = \"scripts\",\n",
    "    arguments = [\n",
    "        \"--model-data\",\n",
    "        training_step.properties.ModelArtifacts.S3ModelArtifacts,\n",
    "        \"--region\",\n",
    "        region,\n",
    "        \"--endpoint-instance-type\",\n",
    "        \"ml.m5.xlarge\",\n",
    "        \"--endpoint-name\",\n",
    "        \"PdM-SKLearn-Pipeline-Endpoint-ReTraining\",\n",
    "    ],\n",
    ")\n",
    "\n",
    "# Creates the pipeline step\n",
    "deploy_step = ProcessingStep(\n",
    "    name = \"PdM-DeployModel\",\n",
    "    step_args = deploy_args\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define a Pipeline of Parameters, Steps, and Conditions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 180,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.workflow.pipeline import Pipeline\n",
    "\n",
    "pipeline_name = \"Predictive-Maintenance-Pipeline\"\n",
    "pipeline = Pipeline(\n",
    "    name = pipeline_name,\n",
    "    parameters = [\n",
    "        processing_instance_count,\n",
    "        instance_type,\n",
    "        model_approval_status,\n",
    "    ],\n",
    "    steps = [preprocess_step, \n",
    "             feature_store_step, \n",
    "             train_test_split_step, \n",
    "             training_step,\n",
    "             register_step, \n",
    "             deploy_step],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "definition = json.loads(pipeline.definition())\n",
    "definition"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "# Update the pipeline execution\n",
    "pipeline.upsert(role_arn = sagemaker_role)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 183,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Execute the pipeline\n",
    "execution = pipeline.start()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# List status of the pipeline steps\n",
    "execution.list_steps()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### List and Check the Endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "endpoint_info = sagemaker_boto_client.describe_endpoint(EndpointName = \"PdM-SKLearn-Pipeline-Endpoint-ReTraining\")\n",
    "endpoint_info[\"EndpointStatus\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sagemaker_boto_client.list_endpoints(NameContains = \"PdM-SKLearn-Pipeline-Endpoint-ReTraining\")[\n",
    "    \"Endpoints\"\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Model Monitoring - Data Drift"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 193,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:sagemaker.image_uris:Defaulting to the only supported framework/algorithm version: .\n",
      "INFO:sagemaker.image_uris:Ignoring unnecessary instance type: None.\n"
     ]
    }
   ],
   "source": [
    "from sagemaker.model_monitor import DefaultModelMonitor\n",
    "from sagemaker.model_monitor.dataset_format import DatasetFormat\n",
    "\n",
    "test_data_drift_monitor = DefaultModelMonitor(\n",
    "    role = sagemaker_role,\n",
    "    instance_count = 1,\n",
    "    instance_type = \"ml.m5.xlarge\",\n",
    "    volume_size_in_gb = 1,\n",
    "    max_runtime_in_seconds = 360,\n",
    "    sagemaker_session = sagemaker_session,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Suggest Baseline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "\n",
    "baseline_data = \"datasets/train-test/test_data_with_headers.csv\"\n",
    "baseline_results_uri = f\"s3://{bucket}/{prefix}/data/baselining/test-header-data-results\"\n",
    "baseline_job_name = f\"PdM-Baseline-Job-Data-Monitor-{datetime.utcnow():%Y-%m-%d-%H%M}\"\n",
    "\n",
    "test_data_drift_monitor.suggest_baseline(\n",
    "    job_name = baseline_job_name,\n",
    "    baseline_dataset = baseline_data,\n",
    "    dataset_format = DatasetFormat.csv(header = True),\n",
    "    output_s3_uri = baseline_results_uri,\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fetch the latest baselining job\n",
    "baseline_job = test_data_drift_monitor.latest_baselining_job"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Preprocessing Script for monitoring schedule data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 233,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting scripts/datacapture_preprocessing.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile scripts/datacapture_preprocessing.py\n",
    "\n",
    "import json\n",
    "\n",
    "def preprocess_handler(inference_record):\n",
    "    input_data = json.loads(inference_record.endpoint_input.data)\n",
    "    input_data = {f\"feature{str(i).zfill(10)}\": val for i, val in enumerate(input_data)}\n",
    "\n",
    "    output_data = json.loads(inference_record.endpoint_output.data)\n",
    "    output_data = {\"prediction0\": output_data}\n",
    "\n",
    "    print(input_data)\n",
    "    print(type(input_data))\n",
    "    print(output_data)\n",
    "    return {**input_data}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 234,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client.upload_file(\n",
    "    Filename=\"scripts/datacapture_preprocessing.py\", Bucket=bucket, Key=f\"{prefix}/code/datacapture_preprocessing.py\"\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Monitoring Schedule"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 217,
   "metadata": {},
   "outputs": [],
   "source": [
    "preprocessor_path = f\"s3://{bucket}/{prefix}/code/datacapture_preprocessing.py\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "monitor_schedule_name = \"PdM-DataDrift-Monitoring-Schedule-Header-Data\"\n",
    "endpoint_name = \"PdM-SKLearn-Pipeline-Endpoint-ReTraining\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 218,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:sagemaker.model_monitor.model_monitoring:Creating Monitoring Schedule with name: PdM-DataDrift-Monitoring-Schedule-2\n"
     ]
    }
   ],
   "source": [
    "from sagemaker.model_monitor import CronExpressionGenerator\n",
    "\n",
    "test_data_drift_monitor.create_monitoring_schedule(\n",
    "    monitor_schedule_name = monitor_schedule_name,\n",
    "    statistics = test_data_drift_monitor.baseline_statistics(),\n",
    "    record_preprocessor_script = preprocessor_path,\n",
    "    endpoint_input = endpoint_name,\n",
    "    constraints = test_data_drift_monitor.suggested_constraints(),\n",
    "    schedule_cron_expression = CronExpressionGenerator.hourly(),\n",
    "    output_s3_uri = baseline_results_uri,\n",
    "    enable_cloudwatch_metrics = True,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Invoke Endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.predictor import Predictor\n",
    "from sagemaker.serializers import CSVSerializer, JSONSerializer\n",
    "\n",
    "predictor = Predictor(\n",
    "    endpoint_name = endpoint_name, \n",
    "    sagemaker_session = sagemaker_session,\n",
    "    serializer = JSONSerializer()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import time\n",
    "\n",
    "for item in test_data.to_numpy()[100:200]:\n",
    "    item = [item.tolist()]\n",
    "    result = predictor.predict(item)\n",
    "    time.sleep(0.5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check the monitoring schedule Execution status"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(test_data_drift_monitor.list_executions())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_data_drift_monitor.list_executions()[-1].describe()[\"ProcessingJobStatus\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_data_drift_monitor.list_executions()[-1].describe()[\"ExitMessage\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check for the latest violations if any generated\n",
    "violations = test_data_drift_monitor.latest_monitoring_constraint_violations()\n",
    "pd.set_option(\"display.max_colwidth\", -1)\n",
    "constraints_df = pd.io.json.json_normalize(violations.body_dict[\"violations\"])\n",
    "constraints_df.head(10)"
   ]
  }
 ],
 "metadata": {
  "instance_type": "ml.t3.medium",
  "kernelspec": {
   "display_name": "Python 3 (Data Science)",
   "language": "python",
   "name": "python3__SAGEMAKER_INTERNAL__arn:aws:sagemaker:us-east-1:081325390199:image/datascience-1.0"
  },
  "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.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}