python_sdk1/exercise09_ml_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exercise09 : ML Pipeline\n",
    "\n",
    "With AML pipeline, you can create ML workflows for the following purposes.\n",
    "\n",
    "- You can build retraining pipeline for MLOps integration.\n",
    "- You can build batch-scoring pipeline instead of real-time scoring in \"[Exercise08 : Publish as a Web Service](./exercise08_publish_model.ipynb)\".\n",
    "\n",
    "ML pipeline can be invoked by the following methods. \n",
    "\n",
    "- Time-based schedule invocation\n",
    "- On-demand invocation by the published endpoint (REST)\n",
    "- Trigger-based invocation, such as, file change or other combined events (with Azure Event Grid, Azure Logic Apps, etc)\n",
    "\n",
    "In this exercise, we create a simple training pipeline, which returns model metrics in top-level (pipeline's) outputs.\n",
    "\n",
    "*back to [index](https://github.com/tsmatz/azureml-tutorial/)*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Get workspace settings\n",
    "\n",
    "Before starting, you must read your configuration settings.<br>\n",
    "When you involve in CI/CD utilities such as GitHub actions, you can also connect to ML workspace without login UI. (See \"[Exercise01 : Prepare Config Settings](./exercise01_prepare_config.ipynb)\".)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Workspace\n",
    "import azureml.core\n",
    "\n",
    "ws = Workspace.from_config()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Prepare resources"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create compute\n",
    "\n",
    "Create your new AML compute for running pipeline.\n",
    "\n",
    "When the pipeline is invoked, the compute will be started. When the pipeline is completed, this compute will be automatically scaled down to zero."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "found existing: mycluster01\n"
     ]
    }
   ],
   "source": [
    "from azureml.core.compute import ComputeTarget, AmlCompute\n",
    "from azureml.core.compute_target import ComputeTargetException\n",
    "\n",
    "try:\n",
    "    compute_target = ComputeTarget(workspace=ws, name='mycluster01')\n",
    "    print('found existing:', compute_target.name)\n",
    "except ComputeTargetException:\n",
    "    print('creating new.')\n",
    "    compute_config = AmlCompute.provisioning_configuration(\n",
    "        vm_size='Standard_D2_v2',\n",
    "        min_nodes=0,\n",
    "        max_nodes=1)\n",
    "    compute_target = ComputeTarget.create(ws, 'mycluster01', compute_config)\n",
    "    compute_target.wait_for_completion(show_output=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Preapare data\n",
    "\n",
    "Get dataset reference for input data.<br>\n",
    "Run \"[Exercise02 : Prepare Data](./exercise02_prepare_data.ipynb)\" beforehand."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Dataset\n",
    "\n",
    "dataset = Dataset.get_by_name(ws, 'mnist_dataset', version='latest')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{\n",
       "    \"assetId\": \"azureml://locations/eastus/workspaces/9f284df9-d636-40ed-bae1-0303c21d4b4f/environments/test-pipeline-env/versions/1\",\n",
       "    \"databricks\": {\n",
       "        \"eggLibraries\": [],\n",
       "        \"jarLibraries\": [],\n",
       "        \"mavenLibraries\": [],\n",
       "        \"pypiLibraries\": [],\n",
       "        \"rcranLibraries\": []\n",
       "    },\n",
       "    \"docker\": {\n",
       "        \"arguments\": [],\n",
       "        \"baseDockerfile\": null,\n",
       "        \"baseImage\": \"mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04\",\n",
       "        \"baseImageRegistry\": {\n",
       "            \"address\": null,\n",
       "            \"password\": null,\n",
       "            \"registryIdentity\": null,\n",
       "            \"username\": null\n",
       "        },\n",
       "        \"buildContext\": null,\n",
       "        \"enabled\": false,\n",
       "        \"platform\": {\n",
       "            \"architecture\": \"amd64\",\n",
       "            \"os\": \"Linux\"\n",
       "        },\n",
       "        \"sharedVolumes\": true,\n",
       "        \"shmSize\": null\n",
       "    },\n",
       "    \"environmentVariables\": {\n",
       "        \"EXAMPLE_ENV_VAR\": \"EXAMPLE_VALUE\"\n",
       "    },\n",
       "    \"inferencingStackVersion\": null,\n",
       "    \"name\": \"test-pipeline-env\",\n",
       "    \"python\": {\n",
       "        \"baseCondaEnvironment\": null,\n",
       "        \"condaDependencies\": {\n",
       "            \"channels\": [\n",
       "                \"anaconda\",\n",
       "                \"conda-forge\"\n",
       "            ],\n",
       "            \"dependencies\": [\n",
       "                \"python=3.8\",\n",
       "                {\n",
       "                    \"pip\": [\n",
       "                        \"tensorflow==2.10.0\"\n",
       "                    ]\n",
       "                }\n",
       "            ],\n",
       "            \"name\": \"project_environment\"\n",
       "        },\n",
       "        \"condaDependenciesFile\": null,\n",
       "        \"interpreterPath\": \"python\",\n",
       "        \"userManagedDependencies\": false\n",
       "    },\n",
       "    \"r\": null,\n",
       "    \"spark\": {\n",
       "        \"packages\": [],\n",
       "        \"precachePackages\": true,\n",
       "        \"repositories\": []\n",
       "    },\n",
       "    \"version\": \"1\"\n",
       "}"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from azureml.core.conda_dependencies import CondaDependencies\n",
    "from azureml.core.environment import Environment\n",
    "\n",
    "# create environment\n",
    "env = Environment('test-pipeline-env')\n",
    "env.python.conda_dependencies = CondaDependencies.create(\n",
    "    python_version=\"3.8\",\n",
    "    pip_packages=['tensorflow==2.10.0'])\n",
    "env.docker.base_image = 'mcr.microsoft.com/azureml/openmpi4.1.0-ubuntu20.04'\n",
    "# You can also use default CPU image (azureml.core.runconfig.DEFAULT_CPU_IMAGE)\n",
    "\n",
    "# register environment to re-use later\n",
    "env.register(workspace=ws)\n",
    "## # speed up by using the existing environment\n",
    "## env = Environment.get(ws, name='test-remote-gpu-env')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create run config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core.runconfig import RunConfiguration\n",
    "\n",
    "run_config = RunConfiguration()\n",
    "run_config.environment = env"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Create Train Step\n",
    "\n",
    "In this example, I create a pipeline for model training, evaluation, and model registration.<br>\n",
    "In this pipeline, the following steps will be executed.\n",
    "\n",
    "1. The model is trained.\n",
    "2. The model accuracy is evaluated. The model metrics is set as pipeline's output.\n",
    "\n",
    "Each source code will then be saved as follows.\n",
    "\n",
    "- training script ```./pipeline_script/train.py```\n",
    "- evaluation script ```./pipeline_script/evaluate.py```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "script_folder = './pipeline_script'\n",
    "os.makedirs(script_folder, exist_ok=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing pipeline_script/train.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile pipeline_script/train.py\n",
    "import os\n",
    "import argparse\n",
    "import tensorflow as tf\n",
    "\n",
    "# parse arguments\n",
    "parser = argparse.ArgumentParser()\n",
    "parser.add_argument(\n",
    "    \"--data_folder\",\n",
    "    type=str,\n",
    "    default=\"./data\",\n",
    "    help=\"Folder path for input data\")\n",
    "parser.add_argument(\n",
    "    \"--model_folder\",\n",
    "    type=str,\n",
    "    default=\"./outputs\",  # AML experiments outputs folder\n",
    "    help=\"Folder path for model output\")\n",
    "parser.add_argument(\n",
    "    \"--learning_rate\",\n",
    "    type=float,\n",
    "    default=\"0.001\",\n",
    "    help=\"Learning Rate\")\n",
    "parser.add_argument(\n",
    "    \"--first_layer\",\n",
    "    type=int,\n",
    "    default=\"128\",\n",
    "    help=\"Neuron number for the first hidden layer\")\n",
    "parser.add_argument(\n",
    "    \"--second_layer\",\n",
    "    type=int,\n",
    "    default=\"64\",\n",
    "    help=\"Neuron number for the second hidden layer\")\n",
    "parser.add_argument(\n",
    "    \"--epochs_num\",\n",
    "    type=int,\n",
    "    default=\"6\",\n",
    "    help=\"Number of epochs\")\n",
    "FLAGS, unparsed = parser.parse_known_args()\n",
    "\n",
    "# build model\n",
    "model = tf.keras.models.Sequential([\n",
    "    tf.keras.layers.Flatten(input_shape=(28, 28)),\n",
    "    tf.keras.layers.Dense(FLAGS.first_layer, activation=\"relu\"),\n",
    "    tf.keras.layers.Dense(FLAGS.second_layer, activation=\"relu\"),\n",
    "    tf.keras.layers.Dense(10)\n",
    "])\n",
    "model.compile(\n",
    "    optimizer=tf.keras.optimizers.Adam(FLAGS.learning_rate),\n",
    "    loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),\n",
    "    metrics=[tf.keras.metrics.SparseCategoricalAccuracy()],\n",
    ")\n",
    "\n",
    "# run training\n",
    "train_data_path = os.path.join(FLAGS.data_folder, \"train\")\n",
    "train_data = tf.data.experimental.load(train_data_path)\n",
    "model.fit(\n",
    "    train_data.shuffle(1000).batch(128).prefetch(tf.data.AUTOTUNE),\n",
    "    epochs=FLAGS.epochs_num\n",
    ")\n",
    "\n",
    "# save model and variables\n",
    "model_path = os.path.join(FLAGS.model_folder, \"mnist_tf_model\")\n",
    "model.save(model_path)\n",
    "print(\"current working directory : \", os.getcwd())\n",
    "print(\"model folder : \", model_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create train step in pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.steps import PythonScriptStep\n",
    "from azureml.pipeline.core import PipelineData\n",
    "\n",
    "pdata_model_folder = PipelineData(\n",
    "    \"model_folder\",\n",
    "    datastore=ws.get_default_datastore(),\n",
    "    is_directory=True\n",
    ")\n",
    "\n",
    "train_step = PythonScriptStep(\n",
    "    name=\"Train Model\",\n",
    "    script_name='train.py',\n",
    "    source_directory='./pipeline_script',\n",
    "    compute_target=compute_target,\n",
    "    outputs=[pdata_model_folder],\n",
    "    arguments=[\n",
    "        '--data_folder',\n",
    "        dataset.as_mount(),\n",
    "        '--model_folder',\n",
    "        pdata_model_folder,\n",
    "    ],\n",
    "    runconfig=run_config,\n",
    "    allow_reuse=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Create Evaluation Step"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create an evalution script ```pipeline_script/evaluate.py```.\n",
    "\n",
    "In this step, the model is evaluated and model metrics (accuracy and loss) is saved."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing pipeline_script/evaluate.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile pipeline_script/evaluate.py\n",
    "import os\n",
    "import json\n",
    "import argparse\n",
    "import tensorflow as tf\n",
    "\n",
    "parser = argparse.ArgumentParser()\n",
    "parser.add_argument(\n",
    "    '--data_folder',\n",
    "    type=str,\n",
    "    default='./data',\n",
    "    help='Folder path for input data')\n",
    "parser.add_argument(\n",
    "    '--model_folder',\n",
    "    type=str,\n",
    "    default='./model',\n",
    "    help='Folder path for model base dir')\n",
    "parser.add_argument(\n",
    "    '--output_info',\n",
    "    type=str,\n",
    "    default='./output_info',\n",
    "    help='File path for model registration info')\n",
    "FLAGS, unparsed = parser.parse_known_args()\n",
    "\n",
    "# load data\n",
    "test_data_path = os.path.join(FLAGS.data_folder, \"test\")\n",
    "test_data = tf.data.experimental.load(test_data_path)\n",
    "\n",
    "# load model\n",
    "model_folder_path = os.path.join(FLAGS.model_folder, \"mnist_tf_model\")\n",
    "loaded_model = tf.keras.models.load_model(model_folder_path)\n",
    "\n",
    "# evaluate\n",
    "results = loaded_model.evaluate(test_data.batch(128))\n",
    "print(\"Loss: {}, Accuracy: {}\".format(results[0], results[1]))\n",
    "\n",
    "# write result (metrics)\n",
    "output_info = {\n",
    "    \"accuracy\" : float(results[1]),\n",
    "    \"loss\" : float(results[0])\n",
    "}\n",
    "output_json = json.dumps(output_info)\n",
    "f = open(FLAGS.output_info,\"w\")\n",
    "f.write(output_json)\n",
    "f.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create evaluation step in pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.pipeline.steps import PythonScriptStep\n",
    "from azureml.pipeline.core import PipelineData\n",
    "\n",
    "pdata_model_info = PipelineData(\n",
    "    \"model_info\",\n",
    "    datastore=ws.get_default_datastore(),\n",
    "    is_directory=False\n",
    ")\n",
    "\n",
    "eval_step = PythonScriptStep(\n",
    "    name=\"Evaluate Model\",\n",
    "    script_name='evaluate.py',\n",
    "    source_directory='./pipeline_script',\n",
    "    compute_target=compute_target,\n",
    "    inputs=[pdata_model_folder],\n",
    "    outputs=[pdata_model_info],\n",
    "    arguments=[\n",
    "        '--data_folder',\n",
    "        dataset.as_mount(),\n",
    "        '--model_folder',\n",
    "        pdata_model_folder,\n",
    "        '--output_info',\n",
    "        pdata_model_info\n",
    "    ],\n",
    "    runconfig=run_config,\n",
    "    allow_reuse=False,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. Create and publish ML pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Step Train Model is ready to be created [c35d4cca]Step Evaluate Model is ready to be created [a193c662]\n",
      "\n",
      "Created step Train Model [c35d4cca][337deae3-d7c2-4ec9-9ce2-d65e38ae68e9], (This step will run and generate new outputs)Created step Evaluate Model [a193c662][73939d6f-d1cc-4b76-b72f-f3cd396c6fc7], (This step will run and generate new outputs)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from azureml.pipeline.core import Pipeline\n",
    "import uuid\n",
    "\n",
    "train_pipeline = Pipeline(workspace=ws, steps=[train_step, eval_step])\n",
    "train_pipeline._set_experiment_name\n",
    "train_pipeline.validate()\n",
    "published_pipeline = train_pipeline.publish(\n",
    "    name=\"training-pipeline01\",\n",
    "    description=\"Model training/evaluation\",\n",
    "    version=str(uuid.uuid4()),\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. Run ML pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When integrating with CI/CD tools, you can submit a new run of this publised pipeline using REST endpoint on demand."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'https://eastus.api.azureml.ms/pipelines/v1.0/subscriptions/b3ae1c15-4fef-4362-8c3a-5d804cdeb18d/resourceGroups/rg-AML/providers/Microsoft.MachineLearningServices/workspaces/ws01/PipelineRuns/PipelineSubmit/d4cffd52-48cb-4e4d-a0f8-e2a99721f6ad'"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# See endpoint url\n",
    "published_pipeline.endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's submit a new run using Python AML SDK."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Submitted PipelineRun 4157af11-73e3-4663-808c-70236f33bdde\n",
      "Link to Azure Machine Learning Portal: https://ml.azure.com/runs/4157af11-73e3-4663-808c-70236f33bdde?wsid=/subscriptions/b3ae1c15-4fef-4362-8c3a-5d804cdeb18d/resourcegroups/rg-AML/workspaces/ws01&tid=72f988bf-86f1-41af-91ab-2d7cd011db47\n"
     ]
    }
   ],
   "source": [
    "from azureml.core import Experiment\n",
    "\n",
    "exp = Experiment(workspace=ws, name='pipeline_experiment01')\n",
    "pipeline_run = exp.submit(published_pipeline)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Go to [AML studio UI](https://ml.azure.com/) and see the progress and results.\n",
    "\n",
    "![Pipeline results](https://tsmatz.github.io/images/github/azure-ml-tensorflow-complete-sample/20220225_Experiment_Pipeline.jpg)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 8. Remove Compute"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Delete cluster (nodes) and remove from AML workspace\n",
    "mycompute = AmlCompute(workspace=ws, name='mycluster01')\n",
    "mycompute.delete()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}