Cloud Pak for Data/Batch Support/Watson OpenScale and Batch Support with IBM Analytics Engine.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"https://github.com/pmservice/ai-openscale-tutorials/raw/master/notebooks/images/banner.png\" align=\"left\" alt=\"banner\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# IBM Watson OpenScale and Batch Processing:<br>Apache Spark on Cloud Pak for Data with IBM Analytics Engine"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook must be run in the Python 3.x runtime environment. It requires Watson OpenScale service credentials.\n",
    "\n",
    "The notebook configures Watson OpenScale to monitor the German Credit Risk model. Use the notebook to enable quality and drift monitoring and run on-demand evaluations. Before you can run the notebook, you must have the following resources:\n",
    "\n",
    "1. The common configuration JSON and Drift Configuration archive generated by using the [common configuration notebook](https://github.com/IBM/watson-openscale-samples/blob/main/Cloud%20Pak%20for%20Data/Batch%20Support/Configuration%20generation%20for%20OpenScale%20batch%20subscription.ipynb).\n",
    "2. Feedback, payload, and drifted transactions tables in an Apache Hive data warehouse that use the data description language statements (DDLs) that are generated as part of running the previous common configuration notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Contents\n",
    "\n",
    "* [1. Setup](#setup)\n",
    "* [2. Configure Watson OpenScale](#openscale)\n",
    "* [3. Set up Subscription](#subscription)\n",
    "* [4. Quality monitoring](#quality)\n",
    "* [5. Drift monitoring](#drift)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. Setup <a name=\"setup\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Package installation\n",
    "\n",
    "First import some of the packages you need to use. After you finish installing the following software packages, restart the kernel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "%env PIP_DISABLE_PIP_VERSION_CHECK=1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install ibm-watson-openscale --no-cache | tail -n 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip show ibm-watson-openscale"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Configure credentials\n",
    "\n",
    "Provide your IBM Watson OpenScale credentials in the following cell:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "WOS_CREDENTIALS = {\n",
    "    \"url\": \"<cluster-url>\",\n",
    "    \"username\": \"<username>\",\n",
    "    \"password\": \"<password>\"\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Specify model details"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Service provider and subscription metadata"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Service Provider\n",
    "\n",
    "SERVICE_PROVIDER_NAME = \"<service-provider-name>\"\n",
    "SERVICE_PROVIDER_DESCRIPTION = \"<service-provider-description>\"\n",
    "\n",
    "# Subscription\n",
    "\n",
    "SUBSCRIPTION_NAME = \"<subscription-name>\"\n",
    "SUBSCRIPTION_DESCRIPTION = \"<subscription-description>\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### IBM Analytics Engine - Spark\n",
    "\n",
    "Make sure that the Apache Spark manager on IBM Analytics Engine is running, and then provide the following details:\n",
    "\n",
    "- IAE_SPARK_DISPLAY_NAME: _Display Name of the Spark instance in IBM Analytics Engine_\n",
    "- IAE_SPARK_JOBS_ENDPOINT: _Spark Jobs Endpoint for IBM Analytics Engine_\n",
    "- IBM_CPD_VOLUME: _IBM Cloud Pak for Data storage volume name_\n",
    "- IBM_CPD_USERNAME: _IBM Cloud Pak for Data username_\n",
    "- IBM_CPD_APIKEY: _IBM Cloud Pak for Data API key_\n",
    "- IAE_SPARK_DESCRIPTION: _Custom description for the Spark instance_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "IAE_SPARK_DISPLAY_NAME = \"<spark-engine-name>\"\n",
    "IAE_SPARK_JOBS_ENDPOINT = \"<spark-job-endpoint-for-ibm-analytics-engine>\"\n",
    "IBM_CPD_VOLUME = \"<ibm-cpd-volume>\"\n",
    "IBM_CPD_USERNAME = \"<ibm-cloud-pak-for-data-username>\"\n",
    "IBM_CPD_APIKEY = \"<ibm-cloud-pak-for-data-apikey>\"\n",
    "IAE_SPARK_DESCRIPTION = \"<iae-spark-description>\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Provide Spark Resource Settings\n",
    "\n",
    "To configure how much of your Spark Cluster resources this job can consume, edit the following values:\n",
    "\n",
    "- max_num_executors: _Maximum Number of executors to launch for this session_\n",
    "- min_executors: _Minimum Number of executors to launch for this session_\n",
    "- executor_cores: _Number of cores to use for each executor_   \n",
    "- executor_memory: _Amount of memory (in GBs) to use per executor process_\n",
    "- driver_cores: _Number of cores to use for the driver process_\n",
    "- driver_memory: _Amount of memory (in GBs) to use for the driver process_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark_parameters = {\n",
    "    \"max_num_executors\": 2,\n",
    "    \"min_num_executors\": 1,\n",
    "    \"executor_cores\": 3,\n",
    "    \"executor_memory\": 2,\n",
    "    \"driver_cores\": 2,\n",
    "    \"driver_memory\": 2\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Apache Hive (Non-Kerberized)\n",
    "\n",
    "The Apache Spark manager on IBM Analytics Engine only supports an Apache Hive data warehouse that is not Kerberos-enabled. To connect to Apache Hive, you must provide the following details:\n",
    "\n",
    "- HIVE_CONNECTION_NAME: _Custom display name for the Hive Connection_\n",
    "- HIVE_CONNECTION_DESCRIPTION: _Custom description for the Hive connection_\n",
    "- HIVE_METASTORE_URI: _Thrift URI for Hive Metastore to connect to_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "HIVE_CONNECTION_NAME = \"<hive-connection-name>\"\n",
    "HIVE_CONNECTION_DESCRIPTION = \"<hive-connection-description>\"\n",
    "HIVE_METASTORE_URI = \"<hive-metastore-uri>\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Feedback table metadata\n",
    "\n",
    "The quality monitor stores metadata in the feedback table. To configure the quality monitor, you must provide the following details. To skip quality monitoring, run the following cell to initialize variables with the value of `None`.\n",
    "\n",
    "- FEEDBACK_DATABASE_NAME: _Database name where feedback table is present_\n",
    "- FEEDBACK_SCHEMA_NAME: _Schema name where feedback table is present_\n",
    "- FEEDBACK_TABLE_NAME: _Name of the feedback table_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#feedback\n",
    "\n",
    "FEEDBACK_DATABASE_NAME = None\n",
    "FEEDBACK_SCHEMA_NAME = None\n",
    "FEEDBACK_TABLE_NAME = None"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Payload and drift table metadata\n",
    "\n",
    "The drift monitor stores metadata in the payload and drift tables. To configure the drift monitor, you must provide the following details. To skip drift monitoring, run the following cell to initialize variables with the value of `None`.\n",
    "\n",
    "- PAYLOAD_DATABASE_NAME: _Database name where payload logging table is present_\n",
    "- PAYLOAD_SCHEMA_NAME: _Schema name where payload logging table is present_\n",
    "- PAYLOAD_TABLE_NAME: _Name of the payload logging table_\n",
    "- DRIFT_DATABASE_NAME: _Database name where drifted transactions table is present_\n",
    "- DRIFT_SCHEMA_NAME: _Schema name where drifted transactions table is present_\n",
    "- DRIFT_TABLE_NAME: _Name of the drifted transactions table_"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#payload logging\n",
    "\n",
    "PAYLOAD_DATABASE_NAME = None\n",
    "PAYLOAD_SCHEMA_NAME = None\n",
    "PAYLOAD_TABLE_NAME = None\n",
    "\n",
    "#drift\n",
    "\n",
    "DRIFT_DATABASE_NAME = None\n",
    "DRIFT_SCHEMA_NAME = None\n",
    "DRIFT_TABLE_NAME = None"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Configure Watson OpenScale <a name=\"openscale\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Import the required libraries and set up the Watson OpenScale client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from ibm_cloud_sdk_core.authenticators import CloudPakForDataAuthenticator\n",
    "from ibm_watson_openscale import *\n",
    "from ibm_watson_openscale.supporting_classes.enums import *\n",
    "from ibm_watson_openscale.supporting_classes import *\n",
    "from ibm_watson_openscale.base_classes.watson_open_scale_v2 import *\n",
    "\n",
    "authenticator = CloudPakForDataAuthenticator(\n",
    "        url=WOS_CREDENTIALS[\"url\"],\n",
    "        username=WOS_CREDENTIALS[\"username\"],\n",
    "        password=WOS_CREDENTIALS[\"password\"],\n",
    "        disable_ssl_verification=True\n",
    "    )\n",
    "\n",
    "wos_client = APIClient(authenticator=authenticator, service_url=WOS_CREDENTIALS[\"url\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Display Watson OpenScale datamart details"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wos_client.data_marts.show()\n",
    "data_marts = wos_client.data_marts.list().result.data_marts\n",
    "data_mart_id=data_marts[0].metadata.id"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a service provider"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Delete existing service provider with the same name as provided\n",
    "\n",
    "service_providers = wos_client.service_providers.list().result.service_providers\n",
    "for provider in service_providers:\n",
    "    if provider.entity.name == SERVICE_PROVIDER_NAME:\n",
    "        wos_client.service_providers.delete(service_provider_id=provider.metadata.id)\n",
    "        break"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add Service Provider\n",
    "\n",
    "added_service_provider_result = wos_client.service_providers.add(\n",
    "        name=SERVICE_PROVIDER_NAME,\n",
    "        description=SERVICE_PROVIDER_DESCRIPTION,\n",
    "        service_type=ServiceTypes.CUSTOM_MACHINE_LEARNING,\n",
    "        credentials={},\n",
    "        operational_space_id=\"production\",\n",
    "        background_mode=False\n",
    "    ).result\n",
    "\n",
    "service_provider_id = added_service_provider_result.metadata.id\n",
    "\n",
    "wos_client.service_providers.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "service_provide_details = wos_client.service_providers.get(service_provider_id=service_provider_id).result\n",
    "print(service_provide_details)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create integrated systems for Spark Engine and Hive"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Delete existing spark and hive integrated systems if present\n",
    "\n",
    "integrated_systems = IntegratedSystems(wos_client).list().result.integrated_systems\n",
    "\n",
    "for system in integrated_systems:\n",
    "    if system.entity.name in (IAE_SPARK_DISPLAY_NAME, HIVE_CONNECTION_NAME):\n",
    "        print(\"Deleting integrated system {}\".format(system.entity.name))\n",
    "        IntegratedSystems(wos_client).delete(integrated_system_id=system.metadata.id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Spark Engine"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark_engine_details = IntegratedSystems(wos_client).add(\n",
    "    name=IAE_SPARK_DISPLAY_NAME,\n",
    "    description=IAE_SPARK_DESCRIPTION,\n",
    "    type=\"spark\",\n",
    "    credentials={\n",
    "        \"username\": IBM_CPD_USERNAME,\n",
    "        \"apikey\": IBM_CPD_APIKEY\n",
    "    },\n",
    "    connection={\n",
    "        \"display_name\": IAE_SPARK_DISPLAY_NAME,\n",
    "        \"endpoint\": IAE_SPARK_JOBS_ENDPOINT,\n",
    "        \"volume\": IBM_CPD_VOLUME,\n",
    "        \"location_type\": \"cpd_iae\"\n",
    "    }\n",
    ").result\n",
    "\n",
    "spark_engine_id = spark_engine_details.metadata.id\n",
    "print(spark_engine_details)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Hive"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hive_connection = {}\n",
    "if HIVE_METASTORE_URI is not None:\n",
    "    hive_connection[\"metastore_url\"] = HIVE_METASTORE_URI\n",
    "    hive_connection[\"location_type\"] = \"metastore\"\n",
    "\n",
    "hive_connection_details = IntegratedSystems(wos_client).add(\n",
    "    name=HIVE_CONNECTION_NAME,\n",
    "    description=HIVE_CONNECTION_DESCRIPTION,\n",
    "    type=\"hive\",\n",
    "    credentials={\n",
    "        \n",
    "    },\n",
    "    connection=hive_connection\n",
    ").result\n",
    "\n",
    "hive_connection_id=hive_connection_details.metadata.id\n",
    "print(hive_connection_details)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Set up a subscription<a name=\"subscription\"></a>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Delete an existing subscription with the provided name\n",
    "\n",
    "subscriptions = wos_client.subscriptions.list().result.subscriptions\n",
    "for sub in subscriptions:\n",
    "    if sub.entity.deployment.name == SUBSCRIPTION_NAME:\n",
    "        wos_client.subscriptions.delete(subscription_id=sub.metadata.id)\n",
    "        break\n",
    "\n",
    "# Display all subscriptions\n",
    "wos_client.subscriptions.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Set subscription metadata\n",
    "\n",
    "In the following cell, type a path to the common configuration JSON file that you created by running the [common configuration notebook](https://github.com/IBM/watson-openscale-samples/blob/main/Cloud%20Pak%20for%20Data/Batch%20Support/Configuration%20generation%20for%20OpenScale%20batch%20subscription.ipynb). After you edit the path information, run the cell to set the asset details and properties, the deployment details, the analytics engine details, and to add the required tables as data sources."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import uuid\n",
    "\n",
    "common_configuration = None\n",
    "with open(\"/path/to/dir/containing/common_config.json\", \"r\") as fp:\n",
    "    common_configuration = json.load(fp).get(\"common_configuration\")\n",
    "    if common_configuration is None:\n",
    "        print(\"Please provide the correct path to the common configuration JSON\")\n",
    "    \n",
    "# Set asset details\n",
    "asset = Asset(\n",
    "    asset_id=str(uuid.uuid4()),\n",
    "    url=\"\",\n",
    "    name=SUBSCRIPTION_NAME,\n",
    "    asset_type=AssetTypes.MODEL,\n",
    "    input_data_type=InputDataType.STRUCTURED,\n",
    "    problem_type=ProblemType.BINARY_CLASSIFICATION\n",
    ")\n",
    "\n",
    "# Set deployment details\n",
    "asset_deployment = AssetDeploymentRequest(\n",
    "    deployment_id=str(uuid.uuid4()),\n",
    "    name=SUBSCRIPTION_NAME,\n",
    "    description=SUBSCRIPTION_DESCRIPTION,\n",
    "    deployment_type=\"batch\"\n",
    ")\n",
    "\n",
    "# Set asset properties \n",
    "asset_properties_request = AssetPropertiesRequest(\n",
    "    label_column=common_configuration[\"label_column\"],\n",
    "    probability_fields=[common_configuration[\"probability\"]],\n",
    "    prediction_field=common_configuration[\"prediction\"],\n",
    "    feature_fields=common_configuration[\"feature_columns\"],\n",
    "    categorical_fields=common_configuration[\"categorical_columns\"]\n",
    ")\n",
    "\n",
    "# Set analytics engine details\n",
    "analytics_engine = AnalyticsEngine(\n",
    "    type=\"spark\",\n",
    "    integrated_system_id=spark_engine_id,\n",
    "    parameters = spark_parameters\n",
    ")\n",
    "\n",
    "# Add selected tables as data sources\n",
    "data_sources = []\n",
    "if FEEDBACK_DATABASE_NAME is not None and FEEDBACK_SCHEMA_NAME is not None and FEEDBACK_TABLE_NAME is not None:\n",
    "    feedback_data_source = DataSource(\n",
    "        type=\"feedback\", \n",
    "        database_name=FEEDBACK_DATABASE_NAME, \n",
    "        schema_name=FEEDBACK_SCHEMA_NAME, \n",
    "        table_name=FEEDBACK_TABLE_NAME, \n",
    "        connection=DataSourceConnection(\n",
    "            type=\"hive\", \n",
    "            integrated_system_id=hive_connection_id\n",
    "        )\n",
    "    )\n",
    "    data_sources.append(feedback_data_source)\n",
    "    \n",
    "if PAYLOAD_DATABASE_NAME is not None and PAYLOAD_SCHEMA_NAME is not None and PAYLOAD_TABLE_NAME is not None \\\n",
    "    and DRIFT_DATABASE_NAME is not None and DRIFT_SCHEMA_NAME is not None and DRIFT_TABLE_NAME is not None:\n",
    "    payload_logging_data_source = DataSource(\n",
    "        type=\"payload\", \n",
    "        database_name=PAYLOAD_DATABASE_NAME, \n",
    "        schema_name=PAYLOAD_SCHEMA_NAME, \n",
    "        table_name=PAYLOAD_TABLE_NAME, \n",
    "        connection=DataSourceConnection(\n",
    "            type=\"hive\", \n",
    "            integrated_system_id=hive_connection_id\n",
    "        )\n",
    "    )\n",
    "    \n",
    "    drifted_transactions_table_data_source = DataSource(\n",
    "        type=\"drift\", \n",
    "        database_name=DRIFT_DATABASE_NAME, \n",
    "        schema_name=DRIFT_SCHEMA_NAME, \n",
    "        table_name=DRIFT_TABLE_NAME, \n",
    "        connection=DataSourceConnection(\n",
    "            type=\"hive\", \n",
    "            integrated_system_id=hive_connection_id\n",
    "        )\n",
    "    )\n",
    "    \n",
    "    data_sources.append(payload_logging_data_source)\n",
    "    data_sources.append(drifted_transactions_table_data_source)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Adding the subscription\n",
    "\n",
    "subscription_details = Subscriptions(wos_client).add(\n",
    "    data_mart_id=data_mart_id,\n",
    "    service_provider_id=service_provider_id,\n",
    "    asset=asset,\n",
    "    deployment=asset_deployment,\n",
    "    asset_properties=asset_properties_request,\n",
    "    analytics_engine=analytics_engine,\n",
    "    data_sources=data_sources).result\n",
    "\n",
    "subscription_id = subscription_details.metadata.id\n",
    "print(subscription_details)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Checking subscription status\n",
    "\n",
    "wos_client.subscriptions.get(subscription_id).result.entity.status.state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add training, output, and input data schemas to the subscription\n",
    "\n",
    "training_data_schema_patch_document=[\n",
    "    JsonPatchOperation(op=OperationTypes.REPLACE, path='/asset_properties/training_data_schema', value=common_configuration[\"training_data_schema\"])\n",
    "]\n",
    "\n",
    "input_data_schema_patch_document=[\n",
    "    JsonPatchOperation(op=OperationTypes.REPLACE, path='/asset_properties/input_data_schema', value=common_configuration[\"input_data_schema\"])\n",
    "]\n",
    "\n",
    "output_data_schema_patch_document=[\n",
    "    JsonPatchOperation(op=OperationTypes.REPLACE, path='/asset_properties/output_data_schema', value=common_configuration[\"output_data_schema\"])\n",
    "]\n",
    "\n",
    "wos_client.subscriptions.update(subscription_id=subscription_id, patch_document=training_data_schema_patch_document)\n",
    "wos_client.subscriptions.update(subscription_id=subscription_id, patch_document=input_data_schema_patch_document)\n",
    "wos_client.subscriptions.update(subscription_id=subscription_id, patch_document=output_data_schema_patch_document)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check subscription status\n",
    "\n",
    "wos_client.subscriptions.get(subscription_id).result.entity.status.state"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. Quality monitoring <a name=\"quality\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Enable the quality monitor\n",
    "\n",
    "In the following code cell, default values are set for the quality monitor. You can change the default values by updating the optional `min_feedback_data_size` attribute in the `parameters` dict and set the quality threshold in the `thresholds` list."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "target = Target(\n",
    "    target_type=TargetTypes.SUBSCRIPTION,\n",
    "    target_id=subscription_id\n",
    ")\n",
    "\n",
    "parameters = {\n",
    "    \"min_feedback_data_size\": 2000000\n",
    "}\n",
    "\n",
    "thresholds = [{\n",
    "        \"metric_id\": \"area_under_roc\",\n",
    "        \"type\": \"lower_limit\",\n",
    "        \"value\": 0.8\n",
    "}]\n",
    "\n",
    "quality_monitor_details = wos_client.monitor_instances.create(\n",
    "    data_mart_id=data_mart_id,\n",
    "    monitor_definition_id=wos_client.monitor_definitions.MONITORS.QUALITY.ID,\n",
    "    target=target,\n",
    "    parameters=parameters,\n",
    "    thresholds=thresholds\n",
    ").result\n",
    "\n",
    "quality_monitor_instance_id = quality_monitor_details.metadata.id\n",
    "print(quality_monitor_details)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check monitor instance status"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "quality_status = None\n",
    "\n",
    "while quality_status not in (\"active\", \"error\"):\n",
    "    monitor_instance_details = wos_client.monitor_instances.get(monitor_instance_id=quality_monitor_instance_id).result\n",
    "    quality_status = monitor_instance_details.entity.status.state\n",
    "    if quality_status not in (\"active\", \"error\"):\n",
    "        print(datetime.utcnow().strftime('%H:%M:%S'), quality_status)\n",
    "        time.sleep(30)\n",
    "        \n",
    "print(datetime.utcnow().strftime('%H:%M:%S'), quality_status)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "monitor_instance_details = wos_client.monitor_instances.get(monitor_instance_id=quality_monitor_instance_id).result\n",
    "print(monitor_instance_details)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Run an on-demand evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check Quality monitor instance details\n",
    "\n",
    "monitor_instance_details = wos_client.monitor_instances.get(monitor_instance_id=quality_monitor_instance_id).result\n",
    "print(monitor_instance_details)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Trigger on-demand run\n",
    "\n",
    "monitoring_run_details = wos_client.monitor_instances.run(monitor_instance_id=quality_monitor_instance_id).result\n",
    "monitoring_run_id=monitoring_run_details.metadata.id\n",
    "\n",
    "print(monitoring_run_details)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check run status\n",
    "\n",
    "quality_run_status = None\n",
    "while quality_run_status not in (\"finished\", \"error\"):\n",
    "    monitoring_run_details = wos_client.monitor_instances.get_run_details(monitor_instance_id=quality_monitor_instance_id, monitoring_run_id=monitoring_run_id).result\n",
    "    quality_run_status = monitoring_run_details.entity.status.state\n",
    "    if quality_run_status not in (\"finished\", \"error\"):\n",
    "        print(datetime.utcnow().strftime(\"%H:%M:%S\"), quality_run_status)\n",
    "        time.sleep(30)\n",
    "        \n",
    "print(datetime.utcnow().strftime(\"%H:%M:%S\"), quality_run_status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Display quality metrics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wos_client.monitor_instances.show_metrics(monitor_instance_id=quality_monitor_instance_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 5. Drift monitoring <a name=\"drift\"></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Enable the drift monitor\n",
    "\n",
    "In the following code cell, type a path to the drift configuration tar ball."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wos_client.monitor_instances.upload_drift_model(\n",
    "    model_path=\"/path/to/dir/containing/drift.tar.gz\",\n",
    "    data_mart_id=data_mart_id,\n",
    "    subscription_id=subscription_id\n",
    ").result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the following code cell, default values are set for the drift monitor. You can change the default values by updating the values in the `parameters` section. The `min_samples` parameter controls the number of records that triggers the drift monitor to run. The `drift_threshold` parameter sets the threshold in decimal format for the drift percentage to trigger an alert. The `train_drift_model` parameter controls whether to re-train the model based on the drift analysis."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "target = Target(\n",
    "    target_type=TargetTypes.SUBSCRIPTION,\n",
    "    target_id=subscription_id\n",
    ")\n",
    "\n",
    "parameters = {\n",
    "    \"min_samples\": 5000,\n",
    "    \"drift_threshold\": 0.05,\n",
    "    \"train_drift_model\": False\n",
    "}\n",
    "\n",
    "drift_monitor_details = wos_client.monitor_instances.create(\n",
    "    data_mart_id=data_mart_id,\n",
    "    monitor_definition_id=wos_client.monitor_definitions.MONITORS.DRIFT.ID,\n",
    "    target=target,\n",
    "    parameters=parameters\n",
    ").result\n",
    "\n",
    "drift_monitor_instance_id = drift_monitor_details.metadata.id\n",
    "print(drift_monitor_details)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check monitor instance status"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "drift_status = None\n",
    "\n",
    "while drift_status not in (\"active\", \"error\"):\n",
    "    monitor_instance_details = wos_client.monitor_instances.get(monitor_instance_id=drift_monitor_instance_id).result\n",
    "    drift_status = monitor_instance_details.entity.status.state\n",
    "    if drift_status not in (\"active\", \"error\"):\n",
    "        print(datetime.utcnow().strftime('%H:%M:%S'), drift_status)\n",
    "        time.sleep(30)\n",
    "\n",
    "print(datetime.utcnow().strftime('%H:%M:%S'), drift_status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Run an on-demand evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check Drift monitor instance details\n",
    "\n",
    "monitor_instance_details = wos_client.monitor_instances.get(monitor_instance_id=drift_monitor_instance_id).result\n",
    "print(monitor_instance_details)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Trigger on-demand run\n",
    "\n",
    "monitoring_run_details = wos_client.monitor_instances.run(monitor_instance_id=drift_monitor_instance_id).result\n",
    "monitoring_run_id=monitoring_run_details.metadata.id\n",
    "\n",
    "print(monitoring_run_details)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check run status\n",
    "\n",
    "drift_run_status = None\n",
    "while drift_run_status not in (\"finished\", \"error\"):\n",
    "    monitoring_run_details = wos_client.monitor_instances.get_run_details(monitor_instance_id=drift_monitor_instance_id, monitoring_run_id=monitoring_run_id).result\n",
    "    drift_run_status = monitoring_run_details.entity.status.state\n",
    "    if drift_run_status not in (\"finished\", \"error\"):\n",
    "        print(datetime.utcnow().strftime(\"%H:%M:%S\"), drift_run_status)\n",
    "        time.sleep(30)\n",
    "        \n",
    "print(datetime.utcnow().strftime(\"%H:%M:%S\"), drift_run_status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Display drift metrics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wos_client.monitor_instances.show_metrics(monitor_instance_id=drift_monitor_instance_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Congratulations!\n",
    "\n",
    "You have finished the Batch demo for IBM Watson OpenScale by using Apache Spark on Cloud Pak for Data IBM Analytics Engine. You can now view the [Watson OpenScale Dashboard](https://url-to-your-cp4d-cluster/aiopenscale). Click the tile for the **German Credit model** to see the quality and drift monitors. Click the timeseries graph to get detailed information on transactions during a specific time window."
   ]
  }
 ],
 "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.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}