sdk/python/endpoints/batch/deploy-models/mnist-classifier/mnist-batch.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Create and manage a batch endpoint for inferencing\n",
    "\n",
    "**Motivations** - In this example, we're going to deploy a model to solve the classic MNIST (\"Modified National Institute of Standards and Technology\") digit recognition problem to perform batch inferencing over large amounts of data (image files). In the first section of this tutorial, we're going to create a batch deployment with a model created using Torch. Such deployment will become our default one in the endpoint. In the second half, we're going to see how we can create a second deployment using a model created with TensorFlow (Keras), test it out, and then switch the endpoint to start using the new deployment as default once we confirm it is working.. "
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 1. Connect to Azure Machine Learning Workspace\n",
    "\n",
    "The [workspace](https://docs.microsoft.com/en-us/azure/machine-learning/concept-workspace) is the top-level resource for Azure Machine Learning, providing a centralized place to work with all the artifacts you create when you use Azure Machine Learning. In this section we will connect to the workspace in which the job will be run.\n",
    "\n",
    "## 1.1. Import the required libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1634854439863
    }
   },
   "outputs": [],
   "source": [
    "from azure.ai.ml import MLClient, Input\n",
    "from azure.ai.ml.entities import (\n",
    "    BatchEndpoint,\n",
    "    ModelBatchDeployment,\n",
    "    ModelBatchDeploymentSettings,\n",
    "    Model,\n",
    "    Environment,\n",
    "    AmlCompute,\n",
    "    BatchRetrySettings,\n",
    "    CodeConfiguration,\n",
    ")\n",
    "from azure.identity import DefaultAzureCredential\n",
    "from azure.ai.ml.constants import AssetTypes, BatchDeploymentOutputAction"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1.2. Configure workspace details and get a handle to the workspace\n",
    "\n",
    "To connect to a workspace, we need identifier parameters - a subscription, resource group and workspace name. We will use these details in the `MLClient` from `azure.ai.ml` to get a handle to the required Azure Machine Learning workspace. We use the default [default azure authentication](https://docs.microsoft.com/en-us/python/api/azure-identity/azure.identity.defaultazurecredential?view=azure-python) for this tutorial. Check the [configuration notebook](../../jobs/configuration.ipynb) for more details on how to configure credentials and connect to a workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1634854439994
    }
   },
   "outputs": [],
   "source": [
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "workspace = \"<AML_WORKSPACE_NAME>\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1634854440154
    }
   },
   "outputs": [],
   "source": [
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you are working in a Azure Machine Learning compute, you can simply:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ml_client = MLClient.from_config(DefaultAzureCredential())"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2. Create Batch Endpoint\n",
    "Batch endpoints are endpoints that are used batch inferencing on large volumes of data over a period of time. Batch endpoints receive pointers to data and run jobs asynchronously to process the data in parallel on compute clusters. Batch endpoints store outputs to a data store for further analysis.\n",
    "\n",
    "To create an online endpoint we will use `BatchEndpoint`. This class allows user to configure the following key aspects:\n",
    "- `name` - Name of the endpoint. Needs to be unique at the Azure region level\n",
    "- `auth_mode` - The authentication method for the endpoint. Currently only Azure Active Directory (Azure AD) token-based (`aad_token`) authentication is supported. \n",
    "- `description`- Description of the endpoint.\n",
    "\n",
    "## 2.1 Configure the endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "name_endpoint"
   },
   "outputs": [],
   "source": [
    "endpoint_name = \"mnist-batch\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1634854440714
    }
   },
   "outputs": [],
   "source": [
    "import random\n",
    "import string\n",
    "\n",
    "# Creating a unique endpoint name by including a random suffix\n",
    "allowed_chars = string.ascii_lowercase + string.digits\n",
    "endpoint_suffix = \"\".join(random.choice(allowed_chars) for x in range(5))\n",
    "endpoint_name = f\"{endpoint_name}-{endpoint_suffix}\"\n",
    "\n",
    "print(f\"Endpoint name: {endpoint_name}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's configure the endpoint:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "configure_endpoint"
   },
   "outputs": [],
   "source": [
    "endpoint = BatchEndpoint(\n",
    "    name=endpoint_name,\n",
    "    description=\"A batch endpoint for scoring images from the MNIST dataset.\",\n",
    "    tags={\"type\": \"deep-learning\"},\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2.2 Create the endpoint\n",
    "Using the `MLClient` created earlier, we will now create the Endpoint in the workspace. This command will start the endpoint creation and return a confirmation response while the endpoint creation continues."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "create_endpoint"
   },
   "outputs": [],
   "source": [
    "ml_client.begin_create_or_update(endpoint).result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Registering the model\n",
    "\n",
    "### 3.1 About the model\n",
    "\n",
    "We are going to deploy a model created using Torch to solve the typical MNIST classification problem."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.2 Registering the model in the workspace\n",
    "\n",
    "We need to register the model in order to use it with Azure Machine Learning:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "register_model"
   },
   "outputs": [],
   "source": [
    "model_name = \"mnist-classifier-torch\"\n",
    "model_local_path = \"deployment-torch/model/\"\n",
    "\n",
    "model = ml_client.models.create_or_update(\n",
    "    Model(\n",
    "        name=model_name,\n",
    "        path=model_local_path,\n",
    "        type=AssetTypes.CUSTOM_MODEL,\n",
    "        tags={\"task\": \"classification\", \"framework\": \"torch\"},\n",
    "    )\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's get a reference to the model:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "get_model"
   },
   "outputs": [],
   "source": [
    "model = ml_client.models.get(name=model_name, label=\"latest\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 4. Create a deployment\n",
    "A deployment is a set of resources required for hosting the model that does the actual inferencing."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.1 Creating an scoring script to work with the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile deployment-torch/code/batch_driver.py\n",
    "\n",
    "import os\n",
    "import pandas as pd\n",
    "import torch\n",
    "import torchvision\n",
    "import glob\n",
    "from os.path import basename\n",
    "from mnist_classifier import MnistClassifier\n",
    "from typing import List\n",
    "\n",
    "def init():\n",
    "    global model\n",
    "    global device\n",
    "\n",
    "    # AZUREML_MODEL_DIR is an environment variable created during deployment\n",
    "    # It is the path to the model folder\n",
    "    model_path = os.environ[\"AZUREML_MODEL_DIR\"]\n",
    "    model_file = glob.glob(f\"{model_path}/*/*.pt\")[-1]\n",
    "\n",
    "    model = MnistClassifier()\n",
    "    model.load_state_dict(torch.load(model_file))\n",
    "    model.eval()\n",
    "\n",
    "    device = torch.device(\"cuda:0\" if torch.cuda.is_available() else \"cpu\")\n",
    "\n",
    "def run(mini_batch: List[str]) -> pd.DataFrame:\n",
    "    print(f\"Executing run method over batch of {len(mini_batch)} files.\")\n",
    "\n",
    "    results = []\n",
    "    with torch.no_grad():\n",
    "        for image_path in mini_batch:\n",
    "            image_data = torchvision.io.read_image(image_path).float()\n",
    "            batch_data = image_data.expand(1, -1, -1, -1)\n",
    "            input = batch_data.to(device)\n",
    "\n",
    "            # perform inference\n",
    "            predict_logits = model(input)\n",
    "\n",
    "            # Compute probabilities, classes and labels\n",
    "            predictions = torch.nn.Softmax(dim=-1)(predict_logits)\n",
    "            predicted_prob, predicted_class= torch.max(predictions, axis=-1)\n",
    "            \n",
    "            results.append({ \n",
    "                \"file\": basename(image_path),\n",
    "                \"class\": predicted_class.numpy()[0],\n",
    "                \"probability\": predicted_prob.numpy()[0]\n",
    "            })\n",
    "\n",
    "    return pd.DataFrame(results)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.2 Creating the compute\n",
    "\n",
    "Batch deployments can run on any Azure ML compute that already exists in the workspace. That means that multiple batch deployments can share the same compute infrastructure. In this example, we are going to work on an AzureML compute cluster called `cpu-cluster`. Let's verify the compute exists on the workspace or create it otherwise."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "create_compute"
   },
   "outputs": [],
   "source": [
    "compute_name = \"batch-cluster\"\n",
    "if not any(filter(lambda m: m.name == compute_name, ml_client.compute.list())):\n",
    "    compute_cluster = AmlCompute(\n",
    "        name=compute_name,\n",
    "        description=\"CPU cluster compute\",\n",
    "        min_instances=0,\n",
    "        max_instances=2,\n",
    "    )\n",
    "    ml_client.compute.begin_create_or_update(compute_cluster).result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Compute may take time to be created. Let's wait for it:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from time import sleep\n",
    "\n",
    "print(f\"Waiting for compute {compute_name}\", end=\"\")\n",
    "while ml_client.compute.get(name=compute_name).provisioning_state == \"Creating\":\n",
    "    sleep(1)\n",
    "    print(\".\", end=\"\")\n",
    "\n",
    "print(\" [DONE]\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.3 Creating the environment\n",
    "\n",
    "Let's create the environment. In our case, our model runs on `Torch`. Azure Machine Learning already has an environment with the required software installed, so we can reutilize this environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "configure_environment"
   },
   "outputs": [],
   "source": [
    "env = Environment(\n",
    "    name=\"batch-torch-py38\",\n",
    "    conda_file=\"deployment-torch/environment/conda.yaml\",\n",
    "    image=\"mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.3 Configure the deployment\n",
    "\n",
    "We will create a deployment for our endpoint using the `BatchDeployment` class. This class allows user to configure the following key aspects.\n",
    "- `name` - Name of the deployment.\n",
    "- `endpoint_name` - Name of the endpoint to create the deployment under.\n",
    "- `model` - The model to use for the deployment. This value can be either a reference to an existing versioned model in the workspace or an inline model specification.\n",
    "- `environment` - The environment to use for the deployment. This value can be either a reference to an existing versioned environment in the workspace or an inline environment specification.\n",
    "- `code_path`- Path to the source code directory for scoring the model\n",
    "- `scoring_script` - Relative path to the scoring file in the source code directory\n",
    "- `compute` - Name of the compute target to execute the batch scoring jobs on\n",
    "- `instance_count`- The number of nodes to use for each batch scoring job.\t\t1\n",
    "- `max_concurrency_per_instance`- The maximum number of parallel scoring_script runs per instance.\n",
    "- `mini_batch_size`\t- The number of files the code_configuration.scoring_script can process in one `run()` call.\n",
    "- `retry_settings`- Retry settings for scoring each mini batch.\t\t\n",
    "   - `max_retries`- The maximum number of retries for a failed or timed-out mini batch (default is 3)\n",
    "   - `timeout`- The timeout in seconds for scoring a mini batch (default is 30)\n",
    "- `output_action`- Indicates how the output should be organized in the output file. Allowed values are `append_row` or `summary_only`. Default is `append_row`\n",
    "- `output_file_name`- Name of the batch scoring output file. Default is `predictions.csv`\n",
    "- `environment_variables`- Dictionary of environment variable name-value pairs to set for each batch scoring job.\n",
    "- `logging_level`- The log verbosity level.\tAllowed values are `warning`, `info`, `debug`. Default is `info`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1634854446905
    },
    "name": "configure_deployment"
   },
   "outputs": [],
   "source": [
    "deployment = ModelBatchDeployment(\n",
    "    name=\"mnist-torch-dpl\",\n",
    "    description=\"A deployment using Torch to solve the MNIST classification dataset.\",\n",
    "    endpoint_name=endpoint_name,\n",
    "    model=model,\n",
    "    code_configuration=CodeConfiguration(\n",
    "        code=\"deployment-torch/code/\", scoring_script=\"batch_driver.py\"\n",
    "    ),\n",
    "    environment=env,\n",
    "    compute=compute_name,\n",
    "    settings=ModelBatchDeploymentSettings(\n",
    "        max_concurrency_per_instance=2,\n",
    "        mini_batch_size=10,\n",
    "        instance_count=2,\n",
    "        output_action=BatchDeploymentOutputAction.APPEND_ROW,\n",
    "        output_file_name=\"predictions.csv\",\n",
    "        retry_settings=BatchRetrySettings(max_retries=3, timeout=30),\n",
    "        logging_level=\"info\",\n",
    "    ),\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.4 Create the deployment\n",
    "Using the `MLClient` created earlier, we will now create the deployment in the workspace. This command will start the deployment creation and return a confirmation response while the deployment creation continues."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "create_deployment"
   },
   "outputs": [],
   "source": [
    "ml_client.begin_create_or_update(deployment).result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's update the default deployment name in the endpoint:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "set_default_deployment"
   },
   "outputs": [],
   "source": [
    "endpoint = ml_client.batch_endpoints.get(endpoint_name)\n",
    "endpoint.defaults.deployment_name = deployment.name\n",
    "ml_client.batch_endpoints.begin_create_or_update(endpoint).result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see the details of the deployment as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "query_deployment"
   },
   "outputs": [],
   "source": [
    "ml_client.batch_deployments.get(name=deployment.name, endpoint_name=endpoint.name)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.5 Test the endpoint with sample data\n",
    "Using the `MLClient` created earlier, we will get a handle to the endpoint. The endpoint can be invoked using the `invoke` command with the following parameters:\n",
    "- `name` - Name of the endpoint\n",
    "- `input_path` - Path where input data is present\n",
    "- `deployment_name` - Name of the specific deployment to test in an endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "# Let's sleep for 2 min to ensure all resources are ready. This is only for automation purposes.\n",
    "time.sleep(120)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4.5.1 Invoke the endpoint\n",
    "\n",
    "Let's now invoke the endpoint for batch scoring job:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "start_batch_scoring_job"
   },
   "outputs": [],
   "source": [
    "job = ml_client.batch_endpoints.invoke(\n",
    "    endpoint_name=endpoint_name,\n",
    "    deployment_name=deployment.name,\n",
    "    input=Input(\n",
    "        path=\"https://azuremlexampledata.blob.core.windows.net/data/mnist/sample/\",\n",
    "        type=AssetTypes.URI_FOLDER,\n",
    "    ),\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4.5.2 Get the details of the invoked job\n",
    "Let us get details and logs of the invoked job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "get_job"
   },
   "outputs": [],
   "source": [
    "ml_client.jobs.get(job.name)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can wait for the job to finish using the following code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "stream_job_logs"
   },
   "outputs": [],
   "source": [
    "ml_client.jobs.stream(job.name)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4.5.3 Download the results\n",
    "\n",
    "The deployment creates a child job that executes the scoring. We can get the details of it using the following code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "scoring_job = list(ml_client.jobs.list(parent_job_name=job.name))[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Job name:\", scoring_job.name)\n",
    "print(\"Job status:\", scoring_job.status)\n",
    "print(\n",
    "    \"Job duration:\",\n",
    "    scoring_job.creation_context.last_modified_at\n",
    "    - scoring_job.creation_context.created_at,\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The outputs generated by the deployment job will be placed in an output named `score`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "download_outputs"
   },
   "outputs": [],
   "source": [
    "ml_client.jobs.download(name=scoring_job.name, download_path=\".\", output_name=\"score\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can read this data using pandas library:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "name": "read_outputs",
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "score = pd.read_csv(\n",
    "    \"named-outputs/score/predictions.csv\",\n",
    "    header=None,\n",
    "    names=[\"file\", \"class\"],\n",
    "    sep=\" \",\n",
    ")\n",
    "score"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.6 Override deployment configuration at invocation time"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4.6.1 Output location\n",
    "\n",
    "You can indicate the output path where you want the job to place the results. To do that, let's first find the ID of a data store registered in AzureML. You can only place outputs in data stores:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "get_data_store"
   },
   "outputs": [],
   "source": [
    "batch_ds = ml_client.datastores.get_default()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run the job:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "start_batch_scoring_job_set_output"
   },
   "outputs": [],
   "source": [
    "filename = f\"predictions-{random.randint(0,99999)}.csv\"\n",
    "\n",
    "job = ml_client.batch_endpoints.invoke(\n",
    "    endpoint_name=endpoint_name,\n",
    "    input=Input(\n",
    "        path=\"https://azuremlexampledata.blob.core.windows.net/data/mnist/sample/\",\n",
    "        type=AssetTypes.URI_FOLDER,\n",
    "    ),\n",
    "    params_override=[\n",
    "        {\"output_dataset.datastore_id\": f\"azureml:{batch_ds.id}\"},\n",
    "        {\"output_dataset.path\": f\"/{endpoint_name}/\"},\n",
    "        {\"output_file_name\": filename},\n",
    "    ],\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 4.6.1 Override deployment configuration settings\n",
    "\n",
    "Some other parameters can be override, including `mini_batch_size` and `instance_count`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "start_batch_scoring_job_overwrite"
   },
   "outputs": [],
   "source": [
    "job = ml_client.batch_endpoints.invoke(\n",
    "    endpoint_name=endpoint_name,\n",
    "    input=Input(\n",
    "        path=\"https://azuremlexampledata.blob.core.windows.net/data/mnist/sample/\"\n",
    "    ),\n",
    "    params_override=[{\"mini_batch_size\": \"20\"}, {\"compute.instance_count\": \"5\"}],\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 5. Create another deployment\n",
    "\n",
    "Let's now create a second deployment, but this time we will solve the same problem using a model trained with TensorFlow and Keras."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "register_model_non_default"
   },
   "outputs": [],
   "source": [
    "model_name = \"mnist-classifier-keras\"\n",
    "model_local_path = \"deployment-keras/model/\"\n",
    "\n",
    "model = ml_client.models.create_or_update(\n",
    "    Model(\n",
    "        name=model_name,\n",
    "        path=model_local_path,\n",
    "        type=AssetTypes.CUSTOM_MODEL,\n",
    "        tags={\"task\": \"classification\", \"framework\": \"tensorflow\"},\n",
    "    )\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's get a reference to the model:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "get_model_non_default"
   },
   "outputs": [],
   "source": [
    "model = ml_client.models.get(name=model_name, label=\"latest\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.1 Creating an scoring script to work with the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile deployment-keras/code/batch_driver.py\n",
    "\n",
    "import os\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import tensorflow as tf\n",
    "from typing import List\n",
    "from os.path import basename\n",
    "from PIL import Image\n",
    "from tensorflow.keras.models import load_model\n",
    "\n",
    "\n",
    "def init():\n",
    "    global model\n",
    "\n",
    "    # AZUREML_MODEL_DIR is an environment variable created during deployment\n",
    "    model_path = os.path.join(os.environ[\"AZUREML_MODEL_DIR\"], \"model\")\n",
    "\n",
    "    # load the model\n",
    "    model = load_model(model_path)\n",
    "\n",
    "def run(mini_batch : List[str]) -> pd.DataFrame:\n",
    "    print(f\"Executing run method over batch of {len(mini_batch)} files.\")\n",
    "\n",
    "    results = []\n",
    "    for image_path in mini_batch:\n",
    "        data = Image.open(image_path)\n",
    "        data = np.array(data)\n",
    "        data_batch = tf.expand_dims(data, axis=0)\n",
    "\n",
    "        # perform inference\n",
    "        pred = model.predict(data_batch)\n",
    "\n",
    "        # Compute probabilities, classes and labels\n",
    "        pred_prob = tf.math.reduce_max(tf.math.softmax(pred, axis=-1)).numpy()\n",
    "        pred_class = tf.math.argmax(pred, axis=-1).numpy()\n",
    "\n",
    "        results.append({\n",
    "            \"file\": basename(image_path), \n",
    "            \"class\": pred_class[0],\n",
    "            \"probability\": pred_prob\n",
    "        })\n",
    "\n",
    "    return pd.DataFrame(results)\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.2 Creating the compute\n",
    "\n",
    "Batch deployments can run on any Azure ML compute that already exists in the workspace. That means that multiple batch deployments can share the same compute infrastructure. In this example, we are going to work on an AzureML compute cluster called `cpu-cluster`."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.3 Creating the environment\n",
    "\n",
    "Let's create the environment. In our case, our model runs on `Torch`. Azure Machine Learning already has an environment with the required software installed, so we can reutilize this environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "configure_environment_non_default"
   },
   "outputs": [],
   "source": [
    "env = Environment(\n",
    "    name=\"batch-tensorflow-py38\",\n",
    "    conda_file=\"deployment-keras/environment/conda.yaml\",\n",
    "    image=\"mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04:latest\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.4 Configure the deployment\n",
    "\n",
    "We will create a deployment for our endpoint using the `BatchDeployment` class. This class allows user to configure the following key aspects.\n",
    "- `name` - Name of the deployment.\n",
    "- `endpoint_name` - Name of the endpoint to create the deployment under.\n",
    "- `model` - The model to use for the deployment. This value can be either a reference to an existing versioned model in the workspace or an inline model specification.\n",
    "- `environment` - The environment to use for the deployment. This value can be either a reference to an existing versioned environment in the workspace or an inline environment specification.\n",
    "- `code_path`- Path to the source code directory for scoring the model\n",
    "- `scoring_script` - Relative path to the scoring file in the source code directory\n",
    "- `compute` - Name of the compute target to execute the batch scoring jobs on\n",
    "- `instance_count`- The number of nodes to use for each batch scoring job.\t\t1\n",
    "- `max_concurrency_per_instance`- The maximum number of parallel scoring_script runs per instance.\n",
    "- `mini_batch_size`\t- The number of files the code_configuration.scoring_script can process in one `run()` call.\n",
    "- `retry_settings`- Retry settings for scoring each mini batch.\t\t\n",
    "   - `max_retries`- The maximum number of retries for a failed or timed-out mini batch (default is 3)\n",
    "   - `timeout`- The timeout in seconds for scoring a mini batch (default is 30)\n",
    "- `output_action`- Indicates how the output should be organized in the output file. Allowed values are `append_row` or `summary_only`. Default is `append_row`\n",
    "- `output_file_name`- Name of the batch scoring output file. Default is `predictions.csv`\n",
    "- `environment_variables`- Dictionary of environment variable name-value pairs to set for each batch scoring job.\n",
    "- `logging_level`- The log verbosity level.\tAllowed values are `warning`, `info`, `debug`. Default is `info`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "configure_deployment_non_default"
   },
   "outputs": [],
   "source": [
    "deployment_keras = ModelBatchDeployment(\n",
    "    name=\"mnist-keras-dpl\",\n",
    "    description=\"A deployment using Keras to solve the MNIST classification dataset.\",\n",
    "    endpoint_name=endpoint_name,\n",
    "    model=model,\n",
    "    code_configuration=CodeConfiguration(\n",
    "        code=\"deployment-keras/code/\", scoring_script=\"batch_driver.py\"\n",
    "    ),\n",
    "    environment=env,\n",
    "    compute=compute_name,\n",
    "    settings=ModelBatchDeploymentSettings(\n",
    "        instance_count=2,\n",
    "        max_concurrency_per_instance=2,\n",
    "        mini_batch_size=10,\n",
    "        output_action=BatchDeploymentOutputAction.APPEND_ROW,\n",
    "        output_file_name=\"predictions.csv\",\n",
    "        retry_settings=BatchRetrySettings(max_retries=3, timeout=30),\n",
    "        logging_level=\"info\",\n",
    "    ),\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.5 Create the deployment\n",
    "Using the `MLClient` created earlier, we will now create the deployment in the workspace. This command will start the deployment creation and return a confirmation response while the deployment creation continues."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "create_deployment_non_default"
   },
   "outputs": [],
   "source": [
    "ml_client.begin_create_or_update(deployment_keras).result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.6 Test the endpoint with sample data\n",
    "Using the `MLClient` created earlier, we will get a handle to the endpoint. The endpoint can be invoked using the `invoke` command with the following parameters:\n",
    "- `name` - Name of the endpoint\n",
    "- `input_path` - Path where input data is present\n",
    "- `deployment_name` - Name of the specific deployment to test in an endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "# Let's sleep for 2 min to ensure all resources are ready. This is only for automation purposes.\n",
    "time.sleep(120)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 5.6.1 Invoke the endpoint"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's now invoke the endpoint for batch scoring job:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "test_deployment_non_default"
   },
   "outputs": [],
   "source": [
    "job = ml_client.batch_endpoints.invoke(\n",
    "    endpoint_name=endpoint_name,\n",
    "    deployment_name=deployment_keras.name,\n",
    "    input=Input(\n",
    "        path=\"https://azuremlexampledata.blob.core.windows.net/data/mnist/sample/\",\n",
    "        type=AssetTypes.URI_FOLDER,\n",
    "    ),\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 5.6.2 Get the details of the invoked job\n",
    "Let us get details and logs of the invoked job"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "get_job_non_default"
   },
   "outputs": [],
   "source": [
    "ml_client.jobs.get(job.name)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can wait for the job to finish using the following code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "stream_job_logs_non_default"
   },
   "outputs": [],
   "source": [
    "ml_client.jobs.stream(job.name)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 5.6.3 Download the results\n",
    "\n",
    "The deployment creates a child job that executes the scoring. We can get the details of it using the following code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "scoring_job = list(ml_client.jobs.list(parent_job_name=job.name))[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Job name:\", scoring_job.name)\n",
    "print(\"Job status:\", scoring_job.status)\n",
    "print(\n",
    "    \"Job duration:\",\n",
    "    scoring_job.creation_context.last_modified_at\n",
    "    - scoring_job.creation_context.created_at,\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The outputs generated by the deployment job will be placed in an output named `score`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "download_outputs_non_default"
   },
   "outputs": [],
   "source": [
    "ml_client.jobs.download(name=scoring_job.name, download_path=\".\", output_name=\"score\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can read this data using pandas library:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "read_outputs_non_default"
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "score = pd.read_csv(\n",
    "    \"named-outputs/score/predictions.csv\",\n",
    "    header=None,\n",
    "    names=[\"file\", \"class\"],\n",
    "    sep=\" \",\n",
    ")\n",
    "score"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 6. Change the default deployment\n",
    "\n",
    "Now that we know the deployment works, we can switch from one deployment to the new one."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "update_default_deployment"
   },
   "outputs": [],
   "source": [
    "endpoint = ml_client.batch_endpoints.get(endpoint_name)\n",
    "endpoint.defaults.deployment_name = deployment_keras.name\n",
    "ml_client.batch_endpoints.begin_create_or_update(endpoint).result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also delete the old deployment:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "delete_deployment"
   },
   "outputs": [],
   "source": [
    "ml_client.batch_deployments.begin_delete(\n",
    "    endpoint_name=endpoint_name, name=deployment.name\n",
    ").result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 7. Clean up Resources\n",
    "Delete endpoint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "name": "delete_endpoint"
   },
   "outputs": [],
   "source": [
    "ml_client.batch_endpoints.begin_delete(name=endpoint_name)"
   ]
  }
 ],
 "metadata": {
  "description": {
   "description": "Create and test batch endpoint and deployement"
  },
  "kernel_info": {
   "name": "python3-azureml"
  },
  "kernelspec": {
   "display_name": "previews",
   "language": "python",
   "name": "previews"
  },
  "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.8.5"
  },
  "nteract": {
   "version": "nteract-front-end@1.0.0"
  },
  "vscode": {
   "interpreter": {
    "hash": "8d732042e5e620df2ddb4aad7f460808ed1754fa045785bdb47941e58456b253"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}